taosdata / TDengine / #3599

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

push

travis-ci

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

fix: reduce write rows from 30w to 3w

3776 of 278949 branches covered (1.35%)

Branch coverage included in aggregate %.

6012 of 274147 relevant lines covered (2.19%)

1642.73 hits per line

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

0.0
/source/libs/executor/src/executor.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 "executor.h"










17



#include "executorInt.h"










18



#include "operator.h"










19



#include "planner.h"










20



#include "querytask.h"










21



#include "tdatablock.h"










22



#include "tref.h"










23



#include "trpc.h"










24



#include "tudf.h"










25



#include "wal.h"










26












27



#include "storageapi.h"










28












29



static TdThreadOnce initPoolOnce = PTHREAD_ONCE_INIT;










30



int32_t             exchangeObjRefPool = -1;










31












32



static void cleanupRefPool() {




×







33



  int32_t ref = atomic_val_compare_exchange_32(&exchangeObjRefPool, exchangeObjRefPool, 0);




×







34



  taosCloseRef(ref);




×







35



}




×







36












37



static void initRefPool() {




×







38



  exchangeObjRefPool = taosOpenRef(1024, doDestroyExchangeOperatorInfo);




×







39



  (void)atexit(cleanupRefPool);




×







40



}




×







41












42



static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {




×







43



  int32_t code = TSDB_CODE_SUCCESS;




×







44



  int32_t lino = 0;




×







45



  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







46



    if (pOperator->numOfDownstream == 0) {





×







47



      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);





×







48



      return TSDB_CODE_APP_ERROR;




×







49



    }










50












51



    if (pOperator->numOfDownstream > 1) {  // not handle this in join query





×







52



      qError("join not supported for stream block scan, %s" PRIx64, id);





×







53



      return TSDB_CODE_APP_ERROR;




×







54



    }










55



    pOperator->status = OP_NOT_OPENED;




×







56



    return doSetSMABlock(pOperator->pDownstream[0], input, numOfBlocks, type, id);




×







57



  } else {










58



    pOperator->status = OP_NOT_OPENED;




×







59












60



    SStreamScanInfo* pInfo = pOperator->info;




×







61












62



    if (type == STREAM_INPUT__MERGED_SUBMIT) {





×







63



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





×







64



        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));




×







65



        void*        tmp = taosArrayPush(pInfo->pBlockLists, pReq);




×







66



        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







67



      }










68



      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;




×







69



    } else if (type == STREAM_INPUT__DATA_SUBMIT) {





×







70



      void* tmp = taosArrayPush(pInfo->pBlockLists, &input);




×







71



      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







72



      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;




×







73



    } else if (type == STREAM_INPUT__DATA_BLOCK) {





×







74



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





×







75



        SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];




×







76



        SPackedData  tmp = {.pDataBlock = pDataBlock};




×







77



        void*        tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);




×







78



        QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);





×







79



      }










80



      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;




×







81



    } else if (type == STREAM_INPUT__CHECKPOINT) {





×







82



      SPackedData tmp = {.pDataBlock = input};




×







83



      void*       tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);




×







84



      QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);





×







85



      pInfo->blockType = STREAM_INPUT__CHECKPOINT;




×







86



    } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {





×







87



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





×







88



        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));




×







89



        void*        tmp = taosArrayPush(pInfo->pBlockLists, pReq);




×







90



        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







91



      }










92



      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;




×







93



    }










94












95



    return TSDB_CODE_SUCCESS;




×







96



  }










97












98



_end:




×







99



  if (code != TSDB_CODE_SUCCESS) {





×







100



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







101



  }










102



  return code;




×







103



}










104












105



static int32_t doSetStreamOpOpen(SOperatorInfo* pOperator, char* id) {




×







106



  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







107



    if (pOperator->numOfDownstream == 0) {





×







108



      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);





×







109



      return TSDB_CODE_APP_ERROR;




×







110



    }










111












112



    if (pOperator->numOfDownstream > 1) {  // not handle this in join query





×







113



      qError("join not supported for stream block scan, %s" PRIx64, id);





×







114



      return TSDB_CODE_APP_ERROR;




×







115



    }










116












117



    pOperator->status = OP_NOT_OPENED;




×







118



    return doSetStreamOpOpen(pOperator->pDownstream[0], id);




×







119



  }










120



  return 0;




×







121



}










122












123



static void clearStreamBlock(SOperatorInfo* pOperator) {




×







124



  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







125



    if (pOperator->numOfDownstream == 1) {





×







126



      return clearStreamBlock(pOperator->pDownstream[0]);




×







127



    }










128



  } else {










129



    SStreamScanInfo* pInfo = pOperator->info;




×







130



    doClearBufferedBlocks(pInfo);




×







131



  }










132



}










133












134



void qResetTaskInfoCode(qTaskInfo_t tinfo) {




×







135



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







136



  pTaskInfo->code = 0;




×







137



  clearStreamBlock(pTaskInfo->pRoot);




×







138



}




×







139












140



static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type,




×







141



                                const char* id) {










142



  int32_t code = TSDB_CODE_SUCCESS;




×







143



  int32_t lino = 0;




×







144



  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







145



    if (pOperator->numOfDownstream == 0) {





×







146



      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);





×







147



      return TSDB_CODE_APP_ERROR;




×







148



    }










149












150



    if (pOperator->numOfDownstream > 1) {  // not handle this in join query





×







151



      qError("join not supported for stream block scan, %s" PRIx64, id);





×







152



      return TSDB_CODE_APP_ERROR;




×







153



    }










154



    pOperator->status = OP_NOT_OPENED;




×







155



    return doSetStreamBlock(pOperator->pDownstream[0], input, numOfBlocks, type, id);




×







156



  } else {










157



    pOperator->status = OP_NOT_OPENED;




×







158



    SStreamScanInfo* pInfo = pOperator->info;




×







159












160



    qDebug("s-task:%s in this batch, %d blocks need to be processed", id, (int32_t)numOfBlocks);





×







161



    QUERY_CHECK_CONDITION((pInfo->validBlockIndex == 0 && taosArrayGetSize(pInfo->pBlockLists) == 0), code, lino, _end,





×







162



                          TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);










163












164



    if (type == STREAM_INPUT__MERGED_SUBMIT) {





×







165



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





×







166



        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));




×







167



        void*        tmp = taosArrayPush(pInfo->pBlockLists, pReq);




×







168



        QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







169



      }










170












171



      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;




×







172



    } else if (type == STREAM_INPUT__DATA_SUBMIT) {





×







173



      void* tmp = taosArrayPush(pInfo->pBlockLists, input);




×







174



      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







175












176



      pInfo->blockType = STREAM_INPUT__DATA_SUBMIT;




×







177



    } else if (type == STREAM_INPUT__DATA_BLOCK) {





×







178



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





×







179



        SSDataBlock* pDataBlock = &((SSDataBlock*)input)[i];




×







180



        SPackedData  tmp = {.pDataBlock = pDataBlock};




×







181



        void*        tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);




×







182



        QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);





×







183



      }










184












185



      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;




×







186



    } else if (type == STREAM_INPUT__CHECKPOINT_TRIGGER) {





×







187



      SPackedData tmp = {.pDataBlock = input};




×







188



      void*       tmpItem = taosArrayPush(pInfo->pBlockLists, &tmp);




×







189



      QUERY_CHECK_NULL(tmpItem, code, lino, _end, terrno);





×







190












191



      pInfo->blockType = STREAM_INPUT__CHECKPOINT;




×







192



    } else {










193



      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;




×







194



      QUERY_CHECK_CODE(code, lino, _end);





×







195



    }










196












197



    return TSDB_CODE_SUCCESS;




×







198



  }










199












200



_end:




×







201



  if (code != TSDB_CODE_SUCCESS) {





×







202



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







203



  }










204



  return code;




×







205



}










206












207



int32_t doSetTaskId(SOperatorInfo* pOperator, SStorageAPI* pAPI) {




×







208



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×







209



  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







210



    SStreamScanInfo* pStreamScanInfo = pOperator->info;




×







211



    if (pStreamScanInfo->pTableScanOp != NULL) {





×







212



      STableScanInfo* pScanInfo = pStreamScanInfo->pTableScanOp->info;




×







213



      if (pScanInfo->base.dataReader != NULL) {





×







214



        int32_t code = pAPI->tsdReader.tsdSetReaderTaskId(pScanInfo->base.dataReader, pTaskInfo->id.str);




×







215



        if (code) {





×







216



          qError("failed to set reader id for executor, code:%s", tstrerror(code));





×







217



          return code;




×







218



        }










219



      }










220



    }










221



  } else {










222



    return doSetTaskId(pOperator->pDownstream[0], pAPI);




×







223



  }










224












225



  return 0;




×







226



}










227












228



int32_t qSetTaskId(qTaskInfo_t tinfo, uint64_t taskId, uint64_t queryId) {




×







229



  SExecTaskInfo* pTaskInfo = tinfo;




×







230



  pTaskInfo->id.queryId = queryId;




×







231



  buildTaskId(taskId, queryId, pTaskInfo->id.str);




×







232












233



  // set the idstr for tsdbReader










234



  return doSetTaskId(pTaskInfo->pRoot, &pTaskInfo->storageAPI);




×







235



}










236












237



int32_t qSetStreamOpOpen(qTaskInfo_t tinfo) {




×







238



  if (tinfo == NULL) {





×







239



    return TSDB_CODE_APP_ERROR;




×







240



  }










241












242



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







243



  int32_t        code = doSetStreamOpOpen(pTaskInfo->pRoot, GET_TASKID(pTaskInfo));




×







244



  if (code != TSDB_CODE_SUCCESS) {





×







245



    qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));





×







246



  } else {










247



    qDebug("%s set the stream block successfully", GET_TASKID(pTaskInfo));





×







248



  }










249












250



  return code;




×







251



}










252












253



int32_t qSetStreamNotifyInfo(qTaskInfo_t tinfo, int32_t eventTypes, const SSchemaWrapper* pSchemaWrapper,




×







254



                             const char* stbFullName, bool newSubTableRule) {










255



  int32_t code = TSDB_CODE_SUCCESS;




×







256



  SStreamTaskInfo *pStreamInfo = NULL;




×







257












258



  if (tinfo == 0 || eventTypes == 0 || pSchemaWrapper == NULL || stbFullName == NULL) {





×







259



    goto _end;




×







260



  }










261












262



  pStreamInfo = &((SExecTaskInfo*)tinfo)->streamInfo;




×







263



  pStreamInfo->eventTypes = eventTypes;





×







264



  pStreamInfo->notifyResultSchema = tCloneSSchemaWrapper(pSchemaWrapper);




×







265



  if (pStreamInfo->notifyResultSchema == NULL) {





×







266



    code = terrno;




×







267



  }










268



  pStreamInfo->stbFullName = taosStrdup(stbFullName);





×







269



  pStreamInfo->newSubTableRule = newSubTableRule;




×







270












271



_end:




×







272



  return code;




×







273



}










274












275



int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {




×







276



  if (tinfo == NULL) {





×







277



    return TSDB_CODE_APP_ERROR;




×







278



  }










279












280



  if (pBlocks == NULL || numOfBlocks == 0) {





×







281



    return TSDB_CODE_SUCCESS;




×







282



  }










283












284



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







285












286



  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*)pBlocks, numOfBlocks, type, GET_TASKID(pTaskInfo));




×







287



  if (code != TSDB_CODE_SUCCESS) {





×







288



    qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));





×







289



  } else {










290



    qDebug("%s set the stream block successfully", GET_TASKID(pTaskInfo));





×







291



  }










292












293



  return code;




×







294



}










295












296



int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {




×







297



  if (tinfo == NULL) {





×







298



    return TSDB_CODE_APP_ERROR;




×







299



  }










300












301



  if (pBlocks == NULL || numOfBlocks == 0) {





×







302



    return TSDB_CODE_SUCCESS;




×







303



  }










304












305



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







306












307



  int32_t code = doSetSMABlock(pTaskInfo->pRoot, (void*)pBlocks, numOfBlocks, type, GET_TASKID(pTaskInfo));




×







308



  if (code != TSDB_CODE_SUCCESS) {





×







309



    qError("%s failed to set the sma block data", GET_TASKID(pTaskInfo));





×







310



  } else {










311



    qDebug("%s set the sma block successfully", GET_TASKID(pTaskInfo));





×







312



  }










313












314



  return code;




×







315



}










316












317



qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* pReaderHandle, int32_t vgId, int32_t* numOfCols,




×







318



                                     uint64_t id) {










319



  if (msg == NULL) {  // create raw scan





×







320



    SExecTaskInfo* pTaskInfo = NULL;




×







321












322



    int32_t code = doCreateTask(0, id, vgId, OPTR_EXEC_MODEL_QUEUE, &pReaderHandle->api, &pTaskInfo);




×







323



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





×







324



      return NULL;




×







325



    }










326












327



    code = createRawScanOperatorInfo(pReaderHandle, pTaskInfo, &pTaskInfo->pRoot);




×







328



    if (NULL == pTaskInfo->pRoot || code != 0) {





×







329



      taosMemoryFree(pTaskInfo);





×







330



      return NULL;




×







331



    }










332












333



    pTaskInfo->storageAPI = pReaderHandle->api;




×







334



    qDebug("create raw scan task info completed, vgId:%d, %s", vgId, GET_TASKID(pTaskInfo));





×







335



    return pTaskInfo;




×







336



  }










337












338



  SSubplan* pPlan = NULL;




×







339



  int32_t   code = qStringToSubplan(msg, &pPlan);




×







340



  if (code != TSDB_CODE_SUCCESS) {





×







341



    terrno = code;




×







342



    return NULL;




×







343



  }










344












345



  qTaskInfo_t pTaskInfo = NULL;




×







346



  code = qCreateExecTask(pReaderHandle, vgId, 0, pPlan, &pTaskInfo, NULL, 0, NULL, OPTR_EXEC_MODEL_QUEUE);




×







347



  if (code != TSDB_CODE_SUCCESS) {





×







348



    qDestroyTask(pTaskInfo);




×







349



    terrno = code;




×







350



    return NULL;




×







351



  }










352












353



  // extract the number of output columns










354



  SDataBlockDescNode* pDescNode = pPlan->pNode->pOutputDataBlockDesc;




×







355



  *numOfCols = 0;




×







356












357



  SNode* pNode;










358



  FOREACH(pNode, pDescNode->pSlots) {





×







359



    SSlotDescNode* pSlotDesc = (SSlotDescNode*)pNode;




×







360



    if (pSlotDesc->output) {





×







361



      ++(*numOfCols);




×







362



    }










363



  }










364












365



  return pTaskInfo;




×







366



}










367












368



int32_t qCreateStreamExecTaskInfo(qTaskInfo_t* pTaskInfo, void* msg, SReadHandle* readers, int32_t vgId, int32_t taskId) {




×







369



  if (msg == NULL) {





×







370



    return TSDB_CODE_INVALID_PARA;




×







371



  }










372












373



  *pTaskInfo = NULL;




×







374












375



  SSubplan* pPlan = NULL;




×







376



  int32_t   code = qStringToSubplan(msg, &pPlan);




×







377



  if (code != TSDB_CODE_SUCCESS) {





×







378



    return code;




×







379



  }










380












381



  code = qCreateExecTask(readers, vgId, taskId, pPlan, pTaskInfo, NULL, 0, NULL, OPTR_EXEC_MODEL_STREAM);




×







382



  if (code != TSDB_CODE_SUCCESS) {





×







383



    qDestroyTask(*pTaskInfo);




×







384



    return code;




×







385



  }










386












387



  code = qStreamInfoResetTimewindowFilter(*pTaskInfo);




×







388



  if (code != TSDB_CODE_SUCCESS) {





×







389



    qDestroyTask(*pTaskInfo);




×







390



  }










391












392



  return code;




×







393



}










394












395



static int32_t filterUnqualifiedTables(const SStreamScanInfo* pScanInfo, const SArray* tableIdList, const char* idstr,




×







396



                                       SStorageAPI* pAPI, SArray** ppArrayRes) {










397



  int32_t code = TSDB_CODE_SUCCESS;




×







398



  int32_t lino = 0;




×







399



  SArray* qa = taosArrayInit(4, sizeof(tb_uid_t));




×







400



  QUERY_CHECK_NULL(qa, code, lino, _error, terrno);





×







401



  int32_t numOfUids = taosArrayGetSize(tableIdList);




×







402



  if (numOfUids == 0) {





×







403



    (*ppArrayRes) = qa;




×







404



    goto _error;




×







405



  }










406












407



  STableScanInfo* pTableScanInfo = pScanInfo->pTableScanOp->info;




×







408












409



  uint64_t suid = 0;




×







410



  uint64_t uid = 0;




×







411



  int32_t  type = 0;




×







412



  tableListGetSourceTableInfo(pTableScanInfo->base.pTableListInfo, &suid, &uid, &type);




×







413












414



  // let's discard the tables those are not created according to the queried super table.










415



  SMetaReader mr = {0};




×







416



  pAPI->metaReaderFn.initReader(&mr, pScanInfo->readHandle.vnode, META_READER_LOCK, &pAPI->metaFn);




×







417



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





×







418



    uint64_t* id = (uint64_t*)taosArrayGet(tableIdList, i);




×







419



    QUERY_CHECK_NULL(id, code, lino, _end, terrno);





×







420












421



    int32_t code = pAPI->metaReaderFn.getTableEntryByUid(&mr, *id);




×







422



    if (code != TSDB_CODE_SUCCESS) {





×







423



      qError("failed to get table meta, uid:%" PRIu64 " code:%s, %s", *id, tstrerror(terrno), idstr);





×







424



      continue;




×







425



    }










426












427



    tDecoderClear(&mr.coder);




×







428












429



    if (mr.me.type == TSDB_SUPER_TABLE) {





×







430



      continue;




×







431



    } else {










432



      if (type == TSDB_SUPER_TABLE) {





×







433



        // this new created child table does not belong to the scanned super table.










434



        if (mr.me.type != TSDB_CHILD_TABLE || mr.me.ctbEntry.suid != suid) {





×







435



          continue;




×







436



        }










437



      } else {  // ordinary table










438



        // In case that the scanned target table is an ordinary table. When replay the WAL during restore the vnode, we










439



        // should check all newly created ordinary table to make sure that this table isn't the destination table.










440



        if (mr.me.uid != uid) {





×







441



          continue;




×







442



        }










443



      }










444



    }










445












446



    if (pScanInfo->pTagCond != NULL) {





×







447



      bool          qualified = false;




×







448



      STableKeyInfo info = {.groupId = 0, .uid = mr.me.uid};




×







449



      code = isQualifiedTable(&info, pScanInfo->pTagCond, pScanInfo->readHandle.vnode, &qualified, pAPI);




×







450



      if (code != TSDB_CODE_SUCCESS) {





×







451



        qError("failed to filter new table, uid:0x%" PRIx64 ", %s", info.uid, idstr);





×







452



        continue;




×







453



      }










454












455



      if (!qualified) {





×







456



        continue;




×







457



      }










458



    }










459












460



    // handle multiple partition










461



    void* tmp = taosArrayPush(qa, id);




×







462



    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







463



  }










464












465



_end:




×







466












467



  pAPI->metaReaderFn.clearReader(&mr);




×







468



  (*ppArrayRes) = qa;




×







469












470



_error:




×







471



  if (code != TSDB_CODE_SUCCESS) {





×







472



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







473



  }










474



  return code;




×







475



}










476












477



int32_t qUpdateTableListForStreamScanner(qTaskInfo_t tinfo, const SArray* tableIdList, bool isAdd) {




×







478



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







479



  const char*    id = GET_TASKID(pTaskInfo);




×







480



  int32_t        code = 0;




×







481












482



  if (isAdd) {





×







483



    qDebug("try to add %d tables id into query list, %s", (int32_t)taosArrayGetSize(tableIdList), id);





×







484



  }










485












486



  // traverse to the stream scanner node to add this table id










487



  SOperatorInfo* pInfo = NULL;




×







488



  code = extractOperatorInTree(pTaskInfo->pRoot, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pInfo);




×







489



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





×







490



    return code;




×







491



  }










492












493



  SStreamScanInfo* pScanInfo = pInfo->info;




×







494



  if (pInfo->pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) {  // clear meta cache for subscription if tag is changed





×







495



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





×







496



      int64_t*        uid = (int64_t*)taosArrayGet(tableIdList, i);




×







497



      STableScanInfo* pTableScanInfo = pScanInfo->pTableScanOp->info;




×







498



      taosLRUCacheErase(pTableScanInfo->base.metaCache.pTableMetaEntryCache, uid, LONG_BYTES);




×







499



    }










500



  }










501












502



  if (isAdd) {  // add new table id





×







503



    SArray* qa = NULL;




×







504



    code = filterUnqualifiedTables(pScanInfo, tableIdList, id, &pTaskInfo->storageAPI, &qa);




×







505



    if (code != TSDB_CODE_SUCCESS) {





×







506



      taosArrayDestroy(qa);




×







507



      return code;




×







508



    }










509



    int32_t numOfQualifiedTables = taosArrayGetSize(qa);




×







510



    qDebug("%d qualified child tables added into stream scanner, %s", numOfQualifiedTables, id);





×







511



    pTaskInfo->storageAPI.tqReaderFn.tqReaderAddTables(pScanInfo->tqReader, qa);




×







512












513



    bool   assignUid = false;




×







514



    size_t bufLen = (pScanInfo->pGroupTags != NULL) ? getTableTagsBufLen(pScanInfo->pGroupTags) : 0;





×







515



    char*  keyBuf = NULL;




×







516



    if (bufLen > 0) {





×







517



      assignUid = groupbyTbname(pScanInfo->pGroupTags);




×







518



      keyBuf = taosMemoryMalloc(bufLen);





×







519



      if (keyBuf == NULL) {





×







520



        taosArrayDestroy(qa);




×







521



        return terrno;




×







522



      }










523



    }










524












525



    STableListInfo* pTableListInfo = ((STableScanInfo*)pScanInfo->pTableScanOp->info)->base.pTableListInfo;




×







526



    taosWLockLatch(&pTaskInfo->lock);




×







527












528



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





×







529



      uint64_t* uid = taosArrayGet(qa, i);




×







530



      if (!uid) {





×







531



        taosMemoryFree(keyBuf);





×







532



        taosArrayDestroy(qa);




×







533



        taosWUnLockLatch(&pTaskInfo->lock);




×







534



        return terrno;




×







535



      }










536



      STableKeyInfo keyInfo = {.uid = *uid, .groupId = 0};




×







537












538



      if (bufLen > 0) {





×







539



        if (assignUid) {





×







540



          keyInfo.groupId = keyInfo.uid;




×







541



        } else {










542



          code = getGroupIdFromTagsVal(pScanInfo->readHandle.vnode, keyInfo.uid, pScanInfo->pGroupTags, keyBuf,




×







543



                                       &keyInfo.groupId, &pTaskInfo->storageAPI);










544



          if (code != TSDB_CODE_SUCCESS) {





×







545



            taosMemoryFree(keyBuf);





×







546



            taosArrayDestroy(qa);




×







547



            taosWUnLockLatch(&pTaskInfo->lock);




×







548



            return code;




×







549



          }










550



        }










551



      }










552












553



      code = tableListAddTableInfo(pTableListInfo, keyInfo.uid, keyInfo.groupId);




×







554



      if (code != TSDB_CODE_SUCCESS) {





×







555



        taosMemoryFree(keyBuf);





×







556



        taosArrayDestroy(qa);




×







557



        taosWUnLockLatch(&pTaskInfo->lock);




×







558



        return code;




×







559



      }










560



    }










561












562



    taosWUnLockLatch(&pTaskInfo->lock);




×







563



    if (keyBuf != NULL) {





×







564



      taosMemoryFree(keyBuf);





×







565



    }










566












567



    taosArrayDestroy(qa);




×







568



  } else {  // remove the table id in current list










569



    qDebug("%d remove child tables from the stream scanner, %s", (int32_t)taosArrayGetSize(tableIdList), id);





×







570



    taosWLockLatch(&pTaskInfo->lock);




×







571



    pTaskInfo->storageAPI.tqReaderFn.tqReaderRemoveTables(pScanInfo->tqReader, tableIdList);




×







572



    taosWUnLockLatch(&pTaskInfo->lock);




×







573



  }










574












575



  return code;




×







576



}










577












578



int32_t qGetQueryTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, int32_t dbNameBuffLen, char* tableName,




×







579



                                    int32_t tbaleNameBuffLen, int32_t* sversion, int32_t* tversion, int32_t idx,










580



                                    bool* tbGet) {










581



  *tbGet = false;




×







582












583



  if (tinfo == NULL || dbName == NULL || tableName == NULL) {





×







584



    return TSDB_CODE_INVALID_PARA;




×







585



  }










586



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







587












588



  if (taosArrayGetSize(pTaskInfo->schemaInfos) <= idx) {





×







589



    return TSDB_CODE_SUCCESS;




×







590



  }










591












592



  SSchemaInfo* pSchemaInfo = taosArrayGet(pTaskInfo->schemaInfos, idx);




×







593



  if (!pSchemaInfo) {





×







594



    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));





×







595



    return terrno;




×







596



  }










597












598



  *sversion = pSchemaInfo->sw->version;




×







599



  *tversion = pSchemaInfo->tversion;




×







600



  if (pSchemaInfo->dbname) {





×







601



    tstrncpy(dbName, pSchemaInfo->dbname, dbNameBuffLen);




×







602



  } else {










603



    dbName[0] = 0;




×







604



  }










605



  if (pSchemaInfo->tablename) {





×







606



    tstrncpy(tableName, pSchemaInfo->tablename, tbaleNameBuffLen);




×







607



  } else {










608



    tableName[0] = 0;




×







609



  }










610












611



  *tbGet = true;




×







612












613



  return TSDB_CODE_SUCCESS;




×







614



}










615












616



bool qIsDynamicExecTask(qTaskInfo_t tinfo) { return ((SExecTaskInfo*)tinfo)->dynamicTask; }




×







617












618



void destroyOperatorParam(SOperatorParam* pParam) {




×







619



  if (NULL == pParam) {





×







620



    return;




×







621



  }










622












623



  // TODO










624



}










625












626



void qUpdateOperatorParam(qTaskInfo_t tinfo, void* pParam) {




×







627



  destroyOperatorParam(((SExecTaskInfo*)tinfo)->pOpParam);




×







628



  ((SExecTaskInfo*)tinfo)->pOpParam = pParam;




×







629



  ((SExecTaskInfo*)tinfo)->paramSet = false;




×







630



}




×







631












632



int32_t qExecutorInit(void) {




×







633



  taosThreadOnce(&initPoolOnce, initRefPool);




×







634



  return TSDB_CODE_SUCCESS;




×







635



}










636












637



int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, SSubplan* pSubplan,




×







638



                        qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, int8_t compressResult, char* sql,










639



                        EOPTR_EXEC_MODEL model) {










640



  SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;




×







641



  (void)taosThreadOnce(&initPoolOnce, initRefPool);




×







642












643



  qDebug("start to create task, TID:0x%" PRIx64 " QID:0x%" PRIx64 ", vgId:%d", taskId, pSubplan->id.queryId, vgId);





×







644












645



  int32_t code = createExecTaskInfo(pSubplan, pTask, readHandle, taskId, vgId, sql, model);




×







646



  if (code != TSDB_CODE_SUCCESS || NULL == *pTask) {





×







647



    qError("failed to createExecTaskInfo, code: %s", tstrerror(code));





×







648



    goto _error;




×







649



  }










650












651



  if (handle) {





×







652



    SDataSinkMgtCfg cfg = {.maxDataBlockNum = 500, .maxDataBlockNumPerQuery = 50, .compress = compressResult};




×







653



    void*           pSinkManager = NULL;




×







654



    code = dsDataSinkMgtInit(&cfg, &(*pTask)->storageAPI, &pSinkManager);




×







655



    if (code != TSDB_CODE_SUCCESS) {





×







656



      qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str);





×







657



      goto _error;




×







658



    }










659












660



    void* pSinkParam = NULL;




×







661



    code = createDataSinkParam(pSubplan->pDataSink, &pSinkParam, (*pTask), readHandle);




×







662



    if (code != TSDB_CODE_SUCCESS) {





×







663



      qError("failed to createDataSinkParam, vgId:%d, code:%s, %s", vgId, tstrerror(code), (*pTask)->id.str);





×







664



      taosMemoryFree(pSinkManager);





×







665



      goto _error;




×







666



    }










667












668



    SDataSinkNode* pSink = NULL;




×







669



    if (readHandle->localExec) {





×







670



      code = nodesCloneNode((SNode *)pSubplan->pDataSink, (SNode **)&pSink);




×







671



      if (code != TSDB_CODE_SUCCESS) {





×







672



        qError("failed to nodesCloneNode, srcType:%d, code:%s, %s", nodeType(pSubplan->pDataSink), tstrerror(code), (*pTask)->id.str);





×







673



        taosMemoryFree(pSinkManager);





×







674



        goto _error;




×







675



      }










676



    }










677












678



    // pSinkParam has been freed during create sinker.










679



    code = dsCreateDataSinker(pSinkManager, readHandle->localExec ? &pSink : &pSubplan->pDataSink, handle, pSinkParam, (*pTask)->id.str);





×







680



    if (code) {





×







681



      qError("s-task:%s failed to create data sinker, code:%s", (*pTask)->id.str, tstrerror(code));





×







682



    }










683



  }










684












685



  qDebug("subplan task create completed, TID:0x%" PRIx64 " QID:0x%" PRIx64 " code:%s", taskId, pSubplan->id.queryId,





×







686



         tstrerror(code));










687












688



_error:




×







689



  // if failed to add ref for all tables in this query, abort current query










690



  return code;




×







691



}










692












693



static void freeBlock(void* param) {




×







694



  SSDataBlock* pBlock = *(SSDataBlock**)param;




×







695



  blockDataDestroy(pBlock);




×







696



}




×







697












698



int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bool* hasMore, SLocalFetch* pLocal) {




×







699



  int32_t        code = TSDB_CODE_SUCCESS;




×







700



  int32_t        lino = 0;




×







701



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







702



  int64_t        threadId = taosGetSelfPthreadId();




×







703












704



  if (pLocal) {





×







705



    memcpy(&pTaskInfo->localFetch, pLocal, sizeof(*pLocal));




×







706



  }










707












708



  taosArrayClear(pResList);




×







709












710



  int64_t curOwner = 0;




×







711



  if ((curOwner = atomic_val_compare_exchange_64(&pTaskInfo->owner, 0, threadId)) != 0) {





×







712



    qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner);





×







713



    pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;




×







714



    return pTaskInfo->code;




×







715



  }










716












717



  if (pTaskInfo->cost.start == 0) {





×







718



    pTaskInfo->cost.start = taosGetTimestampUs();




×







719



  }










720












721



  if (isTaskKilled(pTaskInfo)) {





×







722



    atomic_store_64(&pTaskInfo->owner, 0);




×







723



    qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));





×







724



    return pTaskInfo->code;




×







725



  }










726












727



  // error occurs, record the error code and return to client










728



  int32_t ret = setjmp(pTaskInfo->env);




×







729



  if (ret != TSDB_CODE_SUCCESS) {





×







730



    pTaskInfo->code = ret;




×







731



    (void)cleanUpUdfs();




×







732












733



    qDebug("%s task abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));





×







734



    atomic_store_64(&pTaskInfo->owner, 0);




×







735












736



    return pTaskInfo->code;




×







737



  }










738












739



  qDebug("%s execTask is launched", GET_TASKID(pTaskInfo));





×







740












741



  int32_t      current = 0;




×







742



  SSDataBlock* pRes = NULL;




×







743



  int64_t      st = taosGetTimestampUs();




×







744












745



  if (pTaskInfo->pOpParam && !pTaskInfo->paramSet) {





×







746



    pTaskInfo->paramSet = true;




×







747



    code = pTaskInfo->pRoot->fpSet.getNextExtFn(pTaskInfo->pRoot, pTaskInfo->pOpParam, &pRes);




×







748



  } else {










749



    code = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot, &pRes);




×







750



  }










751












752



  QUERY_CHECK_CODE(code, lino, _end);





×







753



  code = blockDataCheck(pRes);




×







754



  QUERY_CHECK_CODE(code, lino, _end);





×







755












756



  if (pRes == NULL) {





×







757



    st = taosGetTimestampUs();




×







758



  }










759












760



  int32_t rowsThreshold = pTaskInfo->pSubplan->rowsThreshold;




×







761



  if (!pTaskInfo->pSubplan->dynamicRowThreshold || 4096 <= pTaskInfo->pSubplan->rowsThreshold) {





×







762



    rowsThreshold = 4096;




×







763



  }










764












765



  int32_t blockIndex = 0;




×







766



  while (pRes != NULL) {





×







767



    SSDataBlock* p = NULL;




×







768



    if (blockIndex >= taosArrayGetSize(pTaskInfo->pResultBlockList)) {





×







769



      SSDataBlock* p1 = NULL;




×







770



      code = createOneDataBlock(pRes, true, &p1);




×







771



      QUERY_CHECK_CODE(code, lino, _end);





×







772












773



      void* tmp = taosArrayPush(pTaskInfo->pResultBlockList, &p1);




×







774



      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







775



      p = p1;




×







776



    } else {










777



      void* tmp = taosArrayGet(pTaskInfo->pResultBlockList, blockIndex);




×







778



      QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







779












780



      p = *(SSDataBlock**)tmp;




×







781



      code = copyDataBlock(p, pRes);




×







782



      QUERY_CHECK_CODE(code, lino, _end);





×







783



    }










784












785



    blockIndex += 1;




×







786












787



    current += p->info.rows;




×







788



    QUERY_CHECK_CONDITION((p->info.rows > 0 || p->info.type == STREAM_CHECKPOINT), code, lino, _end,





×







789



                          TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);










790



    void* tmp = taosArrayPush(pResList, &p);




×







791



    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







792












793



    if (current >= rowsThreshold) {





×







794



      break;




×







795



    }










796












797



    code = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot, &pRes);




×







798



    QUERY_CHECK_CODE(code, lino, _end);





×







799



    code = blockDataCheck(pRes);




×







800



    QUERY_CHECK_CODE(code, lino, _end);





×







801



  }










802












803



  if (pTaskInfo->pSubplan->dynamicRowThreshold) {





×







804



    pTaskInfo->pSubplan->rowsThreshold -= current;




×







805



  }










806












807



  *hasMore = (pRes != NULL);




×







808



  uint64_t el = (taosGetTimestampUs() - st);




×







809












810



  pTaskInfo->cost.elapsedTime += el;




×







811



  if (NULL == pRes) {





×







812



    *useconds = pTaskInfo->cost.elapsedTime;




×







813



  }










814












815



_end:




×







816



  (void)cleanUpUdfs();




×







817












818



  uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows;




×







819



  qDebug("%s task suspended, %d rows in %d blocks returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",





×







820



         GET_TASKID(pTaskInfo), current, (int32_t)taosArrayGetSize(pResList), total, 0, el / 1000.0);










821












822



  atomic_store_64(&pTaskInfo->owner, 0);




×







823



  if (code) {





×







824



    pTaskInfo->code = code;




×







825



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







826



  }










827












828



  return pTaskInfo->code;




×







829



}










830












831



void qCleanExecTaskBlockBuf(qTaskInfo_t tinfo) {




×







832



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







833



  SArray*        pList = pTaskInfo->pResultBlockList;




×







834



  size_t         num = taosArrayGetSize(pList);




×







835



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





×







836



    SSDataBlock** p = taosArrayGet(pTaskInfo->pResultBlockList, i);




×







837



    if (p) {





×







838



      blockDataDestroy(*p);




×







839



    }










840



  }










841












842



  taosArrayClear(pTaskInfo->pResultBlockList);




×







843



}




×







844












845



int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) {




×







846



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







847



  int64_t        threadId = taosGetSelfPthreadId();




×







848



  int64_t        curOwner = 0;




×







849












850



  *pRes = NULL;




×







851












852



  // todo extract method










853



  taosRLockLatch(&pTaskInfo->lock);




×







854



  bool isKilled = isTaskKilled(pTaskInfo);




×







855



  if (isKilled) {





×







856



    clearStreamBlock(pTaskInfo->pRoot);




×







857



    qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));





×







858












859



    taosRUnLockLatch(&pTaskInfo->lock);




×







860



    return pTaskInfo->code;




×







861



  }










862












863



  if (pTaskInfo->owner != 0) {





×







864



    qError("%s-%p execTask is now executed by thread:%p", GET_TASKID(pTaskInfo), pTaskInfo, (void*)curOwner);





×







865



    pTaskInfo->code = TSDB_CODE_QRY_IN_EXEC;




×







866












867



    taosRUnLockLatch(&pTaskInfo->lock);




×







868



    return pTaskInfo->code;




×







869



  }










870












871



  pTaskInfo->owner = threadId;




×







872



  taosRUnLockLatch(&pTaskInfo->lock);




×







873












874



  if (pTaskInfo->cost.start == 0) {





×







875



    pTaskInfo->cost.start = taosGetTimestampUs();




×







876



  }










877












878



  // error occurs, record the error code and return to client










879



  int32_t ret = setjmp(pTaskInfo->env);




×







880



  if (ret != TSDB_CODE_SUCCESS) {





×







881



    pTaskInfo->code = ret;




×







882



    (void)cleanUpUdfs();




×







883



    qDebug("%s task abort due to error/cancel occurs, code:%s", GET_TASKID(pTaskInfo), tstrerror(pTaskInfo->code));





×







884



    atomic_store_64(&pTaskInfo->owner, 0);




×







885



    return pTaskInfo->code;




×







886



  }










887












888



  qDebug("%s execTask is launched", GET_TASKID(pTaskInfo));





×







889












890



  int64_t st = taosGetTimestampUs();




×







891












892



  int32_t code = pTaskInfo->pRoot->fpSet.getNextFn(pTaskInfo->pRoot, pRes);




×







893



  if (code) {





×







894



    pTaskInfo->code = code;




×







895



    qError("%s failed at line %d, code:%s %s", __func__, __LINE__, tstrerror(code), GET_TASKID(pTaskInfo));





×







896



  }










897












898



  code = blockDataCheck(*pRes);




×







899



  if (code) {





×







900



    pTaskInfo->code = code;




×







901



    qError("%s failed at line %d, code:%s %s", __func__, __LINE__, tstrerror(code), GET_TASKID(pTaskInfo));





×







902



  }










903












904



  uint64_t el = (taosGetTimestampUs() - st);




×







905












906



  pTaskInfo->cost.elapsedTime += el;




×







907



  if (NULL == *pRes) {





×







908



    *useconds = pTaskInfo->cost.elapsedTime;




×







909



  }










910












911



  (void) cleanUpUdfs();




×







912












913



  int32_t  current = (*pRes != NULL) ? (*pRes)->info.rows : 0;





×







914



  uint64_t total = pTaskInfo->pRoot->resultInfo.totalRows;




×







915












916



  qDebug("%s task suspended, %d rows returned, total:%" PRId64 " rows, in sinkNode:%d, elapsed:%.2f ms",





×







917



         GET_TASKID(pTaskInfo), current, total, 0, el / 1000.0);










918












919



  atomic_store_64(&pTaskInfo->owner, 0);




×







920



  return pTaskInfo->code;




×







921



}










922












923



int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) {




×







924



  taosWLockLatch(&pTaskInfo->stopInfo.lock);




×







925



  void* tmp = taosArrayPush(pTaskInfo->stopInfo.pStopInfo, pInfo);




×







926



  taosWUnLockLatch(&pTaskInfo->stopInfo.lock);




×







927












928



  if (!tmp) {





×







929



    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));





×







930



    return terrno;




×







931



  }










932



  return TSDB_CODE_SUCCESS;




×







933



}










934












935



int32_t stopInfoComp(void const* lp, void const* rp) {




×







936



  SExchangeOpStopInfo* key = (SExchangeOpStopInfo*)lp;




×







937



  SExchangeOpStopInfo* pInfo = (SExchangeOpStopInfo*)rp;




×







938












939



  if (key->refId < pInfo->refId) {





×







940



    return -1;




×







941



  } else if (key->refId > pInfo->refId) {





×







942



    return 1;




×







943



  }










944












945



  return 0;




×







946



}










947












948



void qRemoveTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo) {




×







949



  taosWLockLatch(&pTaskInfo->stopInfo.lock);




×







950



  int32_t idx = taosArraySearchIdx(pTaskInfo->stopInfo.pStopInfo, pInfo, stopInfoComp, TD_EQ);




×







951



  if (idx >= 0) {





×







952



    taosArrayRemove(pTaskInfo->stopInfo.pStopInfo, idx);




×







953



  }










954



  taosWUnLockLatch(&pTaskInfo->stopInfo.lock);




×







955



}




×







956












957



void qStopTaskOperators(SExecTaskInfo* pTaskInfo) {




×







958



  taosWLockLatch(&pTaskInfo->stopInfo.lock);




×







959












960



  int32_t num = taosArrayGetSize(pTaskInfo->stopInfo.pStopInfo);




×







961



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





×







962



    SExchangeOpStopInfo* pStop = taosArrayGet(pTaskInfo->stopInfo.pStopInfo, i);




×







963



    if (!pStop) {





×







964



      qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));





×







965



      continue;




×







966



    }










967



    SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pStop->refId);




×







968



    if (pExchangeInfo) {





×







969



      int32_t code = tsem_post(&pExchangeInfo->ready);




×







970



      if (code != TSDB_CODE_SUCCESS) {





×







971



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







972



      }










973



      code = taosReleaseRef(exchangeObjRefPool, pStop->refId);




×







974



      if (code != TSDB_CODE_SUCCESS) {





×







975



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







976



      }










977



    }










978



  }










979












980



  taosWUnLockLatch(&pTaskInfo->stopInfo.lock);




×







981



}




×







982












983



int32_t qAsyncKillTask(qTaskInfo_t qinfo, int32_t rspCode) {




×







984



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;




×







985



  if (pTaskInfo == NULL) {





×







986



    return TSDB_CODE_QRY_INVALID_QHANDLE;




×







987



  }










988












989



  qDebug("%s execTask async killed", GET_TASKID(pTaskInfo));





×







990












991



  setTaskKilled(pTaskInfo, rspCode);




×







992



  qStopTaskOperators(pTaskInfo);




×







993












994



  return TSDB_CODE_SUCCESS;




×







995



}










996












997



int32_t qKillTask(qTaskInfo_t tinfo, int32_t rspCode) {




×







998



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







999



  if (pTaskInfo == NULL) {





×







1000



    return TSDB_CODE_QRY_INVALID_QHANDLE;




×







1001



  }










1002












1003



  qDebug("%s sync killed execTask", GET_TASKID(pTaskInfo));





×







1004



  setTaskKilled(pTaskInfo, TSDB_CODE_TSC_QUERY_KILLED);




×







1005












1006



  while (1) {










1007



    taosWLockLatch(&pTaskInfo->lock);




×







1008



    if (qTaskIsExecuting(pTaskInfo)) {  // let's wait for 100 ms and try again





×







1009



      taosWUnLockLatch(&pTaskInfo->lock);




×







1010



      taosMsleep(100);




×







1011



    } else {  // not running now










1012



      pTaskInfo->code = rspCode;




×







1013



      taosWUnLockLatch(&pTaskInfo->lock);




×







1014



      return TSDB_CODE_SUCCESS;




×







1015



    }










1016



  }










1017



}










1018












1019



bool qTaskIsExecuting(qTaskInfo_t qinfo) {




×







1020



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;




×







1021



  if (NULL == pTaskInfo) {





×







1022



    return false;




×







1023



  }










1024












1025



  return 0 != atomic_load_64(&pTaskInfo->owner);




×







1026



}










1027












1028



static void printTaskExecCostInLog(SExecTaskInfo* pTaskInfo) {




×







1029



  STaskCostInfo* pSummary = &pTaskInfo->cost;




×







1030



  int64_t        idleTime = pSummary->start - pSummary->created;




×







1031












1032



  SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;




×







1033



  if (pSummary->pRecoder != NULL) {





×







1034



    qDebug(





×







1035



        "%s :cost summary: idle:%.2f ms, elapsed time:%.2f ms, extract tableList:%.2f ms, "










1036



        "createGroupIdMap:%.2f ms, total blocks:%d, "










1037



        "load block SMA:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,










1038



        GET_TASKID(pTaskInfo), idleTime / 1000.0, pSummary->elapsedTime / 1000.0, pSummary->extractListTime,










1039



        pSummary->groupIdMapTime, pRecorder->totalBlocks, pRecorder->loadBlockStatis, pRecorder->loadBlocks,










1040



        pRecorder->totalRows, pRecorder->totalCheckedRows);










1041



  } else {










1042



    qDebug("%s :cost summary: idle in queue:%.2f ms, elapsed time:%.2f ms", GET_TASKID(pTaskInfo), idleTime / 1000.0,





×







1043



           pSummary->elapsedTime / 1000.0);










1044



  }










1045



}




×







1046












1047



void qDestroyTask(qTaskInfo_t qTaskHandle) {




×







1048



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qTaskHandle;




×







1049



  if (pTaskInfo == NULL) {





×







1050



    return;




×







1051



  }










1052












1053



  if (pTaskInfo->pRoot != NULL) {





×







1054



    qDebug("%s execTask completed, numOfRows:%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->pRoot->resultInfo.totalRows);





×







1055



  } else {










1056



    qDebug("%s execTask completed", GET_TASKID(pTaskInfo));





×







1057



  }










1058












1059



  printTaskExecCostInLog(pTaskInfo);  // print the query cost summary




×







1060



  doDestroyTask(pTaskInfo);




×







1061



}










1062












1063



int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, SArray* pExecInfoList) {




×







1064



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1065



  return getOperatorExplainExecInfo(pTaskInfo->pRoot, pExecInfoList);




×







1066



}










1067












1068



void qExtractStreamScanner(qTaskInfo_t tinfo, void** scanner) {




×







1069



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1070



  SOperatorInfo* pOperator = pTaskInfo->pRoot;




×







1071












1072



  while (1) {




×







1073



    uint16_t type = pOperator->operatorType;




×







1074



    if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







1075



      *scanner = pOperator->info;




×







1076



      break;




×







1077



    } else {










1078



      pOperator = pOperator->pDownstream[0];




×







1079



    }










1080



  }










1081



}




×







1082












1083



int32_t qStreamSourceScanParamForHistoryScanStep1(qTaskInfo_t tinfo, SVersionRange* pVerRange, STimeWindow* pWindow) {




×







1084



  int32_t        code = TSDB_CODE_SUCCESS;




×







1085



  int32_t        lino = 0;




×







1086



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1087



  QUERY_CHECK_CONDITION((pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM), code, lino, _end,





×







1088



                        TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);










1089












1090



  SStreamTaskInfo* pStreamInfo = &pTaskInfo->streamInfo;




×







1091












1092



  pStreamInfo->fillHistoryVer = *pVerRange;




×







1093



  pStreamInfo->fillHistoryWindow = *pWindow;




×







1094



  pStreamInfo->recoverStep = STREAM_RECOVER_STEP__PREPARE1;




×







1095












1096



  qDebug("%s step 1. set param for stream scanner for scan-history data, verRange:%" PRId64 " - %" PRId64





×







1097



         ", window:%" PRId64 " - %" PRId64,










1098



         GET_TASKID(pTaskInfo), pStreamInfo->fillHistoryVer.minVer, pStreamInfo->fillHistoryVer.maxVer, pWindow->skey,










1099



         pWindow->ekey);










1100



_end:




×







1101



  if (code != TSDB_CODE_SUCCESS) {





×







1102



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







1103



  }










1104



  return code;




×







1105



}










1106












1107



int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRange* pVerRange, STimeWindow* pWindow) {




×







1108



  int32_t        code = TSDB_CODE_SUCCESS;




×







1109



  int32_t        lino = 0;




×







1110



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1111



  QUERY_CHECK_CONDITION((pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM), code, lino, _end,





×







1112



                        TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);










1113












1114



  SStreamTaskInfo* pStreamInfo = &pTaskInfo->streamInfo;




×







1115












1116



  pStreamInfo->fillHistoryVer = *pVerRange;




×







1117



  pStreamInfo->fillHistoryWindow = *pWindow;




×







1118



  pStreamInfo->recoverStep = STREAM_RECOVER_STEP__PREPARE2;




×







1119












1120



  qDebug("%s step 2. set param for stream scanner scan wal, verRange:%" PRId64 "-%" PRId64 ", window:%" PRId64





×







1121



         "-%" PRId64,










1122



         GET_TASKID(pTaskInfo), pStreamInfo->fillHistoryVer.minVer, pStreamInfo->fillHistoryVer.maxVer, pWindow->skey,










1123



         pWindow->ekey);










1124



_end:




×







1125



  if (code != TSDB_CODE_SUCCESS) {





×







1126



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







1127



  }










1128



  return code;




×







1129



}










1130












1131



int32_t qStreamRecoverFinish(qTaskInfo_t tinfo) {




×







1132



  int32_t        code = TSDB_CODE_SUCCESS;




×







1133



  int32_t        lino = 0;




×







1134



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1135



  QUERY_CHECK_CONDITION((pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM), code, lino, _end,





×







1136



                        TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);










1137



  pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;




×







1138












1139



_end:




×







1140



  if (code != TSDB_CODE_SUCCESS) {





×







1141



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







1142



  }










1143



  return code;




×







1144



}










1145












1146



static int32_t getOpratorIntervalInfo(SOperatorInfo* pOperator, int64_t* pWaterMark, SInterval* pInterval, STimeWindow* pLastWindow) {




×







1147



  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







1148



    return getOpratorIntervalInfo(pOperator->pDownstream[0], pWaterMark, pInterval, pLastWindow);




×







1149



  }










1150



  SStreamScanInfo* pScanOp = (SStreamScanInfo*) pOperator->info;




×







1151



  *pWaterMark = pScanOp->twAggSup.waterMark;




×







1152



  *pInterval = pScanOp->interval;




×







1153



  *pLastWindow = pScanOp->lastScanRange;




×







1154



  return TSDB_CODE_SUCCESS; 




×







1155



}










1156












1157



int32_t qGetStreamIntervalExecInfo(qTaskInfo_t tinfo, int64_t* pWaterMark, SInterval* pInterval, STimeWindow* pLastWindow) {




×







1158



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1159



  SOperatorInfo* pOperator = pTaskInfo->pRoot;




×







1160



  return getOpratorIntervalInfo(pOperator, pWaterMark, pInterval, pLastWindow);




×







1161



}










1162












1163



int32_t qSetStreamOperatorOptionForScanHistory(qTaskInfo_t tinfo) {




×







1164



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1165



  SOperatorInfo* pOperator = pTaskInfo->pRoot;




×







1166












1167



  while (1) {




×







1168



    int32_t type = pOperator->operatorType;




×







1169



    if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL || type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||





×







1170



        type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL ||





×







1171



        type == QUERY_NODE_PHYSICAL_PLAN_STREAM_MID_INTERVAL) {




×







1172



      SStreamIntervalOperatorInfo* pInfo = pOperator->info;




×







1173



      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;




×







1174












1175



      qInfo("save stream param for interval: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);





×







1176












1177



      pSup->calTriggerSaved = pSup->calTrigger;




×







1178



      pSup->deleteMarkSaved = pSup->deleteMark;




×







1179



      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;




×







1180



      pSup->deleteMark = INT64_MAX;




×







1181



      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;




×







1182



      pInfo->ignoreExpiredData = false;




×







1183



    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||





×







1184



               type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||





×







1185



               type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {




×







1186



      SStreamSessionAggOperatorInfo* pInfo = pOperator->info;




×







1187



      STimeWindowAggSupp*            pSup = &pInfo->twAggSup;




×







1188












1189



      qInfo("save stream param for session: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);





×







1190












1191



      pSup->calTriggerSaved = pSup->calTrigger;




×







1192



      pSup->deleteMarkSaved = pSup->deleteMark;




×







1193



      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;




×







1194



      pSup->deleteMark = INT64_MAX;




×







1195



      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;




×







1196



      pInfo->ignoreExpiredData = false;




×







1197



    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {





×







1198



      SStreamStateAggOperatorInfo* pInfo = pOperator->info;




×







1199



      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;




×







1200












1201



      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);





×







1202












1203



      pSup->calTriggerSaved = pSup->calTrigger;




×







1204



      pSup->deleteMarkSaved = pSup->deleteMark;




×







1205



      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;




×







1206



      pSup->deleteMark = INT64_MAX;




×







1207



      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;




×







1208



      pInfo->ignoreExpiredData = false;




×







1209



    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_EVENT) {





×







1210



      SStreamEventAggOperatorInfo* pInfo = pOperator->info;




×







1211



      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;




×







1212












1213



      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);





×







1214












1215



      pSup->calTriggerSaved = pSup->calTrigger;




×







1216



      pSup->deleteMarkSaved = pSup->deleteMark;




×







1217



      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;




×







1218



      pSup->deleteMark = INT64_MAX;




×







1219



      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;




×







1220



      pInfo->ignoreExpiredData = false;




×







1221



    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_COUNT) {





×







1222



      SStreamCountAggOperatorInfo* pInfo = pOperator->info;




×







1223



      STimeWindowAggSupp*          pSup = &pInfo->twAggSup;




×







1224












1225



      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);





×







1226












1227



      pSup->calTriggerSaved = pSup->calTrigger;




×







1228



      pSup->deleteMarkSaved = pSup->deleteMark;




×







1229



      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;




×







1230



      pSup->deleteMark = INT64_MAX;




×







1231



      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;




×







1232



      pInfo->ignoreExpiredData = false;




×







1233



      qInfo("save stream task:%s, param for state: %d", GET_TASKID(pTaskInfo), pInfo->ignoreExpiredData);





×







1234



    } else if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERP_FUNC) {





×







1235



      SStreamTimeSliceOperatorInfo* pInfo = pOperator->info;




×







1236



      STimeWindowAggSupp*           pSup = &pInfo->twAggSup;




×







1237












1238



      qInfo("save stream param for state: %d,  %" PRId64, pSup->calTrigger, pSup->deleteMark);





×







1239












1240



      pSup->calTriggerSaved = pSup->calTrigger;




×







1241



      pSup->deleteMarkSaved = pSup->deleteMark;




×







1242



      pSup->calTrigger = STREAM_TRIGGER_AT_ONCE;




×







1243



      pSup->deleteMark = INT64_MAX;




×







1244



      pInfo->ignoreExpiredDataSaved = pInfo->ignoreExpiredData;




×







1245



      pInfo->ignoreExpiredData = false;




×







1246



      qInfo("save stream task:%s, param for state: %d", GET_TASKID(pTaskInfo), pInfo->ignoreExpiredData);





×







1247



    }










1248












1249



    // iterate operator tree










1250



    if (pOperator->numOfDownstream != 1 || pOperator->pDownstream[0] == NULL) {





×







1251



      if (pOperator->numOfDownstream > 1) {





×







1252



        qError("unexpected stream, multiple downstream");





×







1253



        return -1;




×







1254



      }










1255



      return 0;




×







1256



    } else {










1257



      pOperator = pOperator->pDownstream[0];




×







1258



    }










1259



  }










1260












1261



  return 0;










1262



}










1263












1264



bool qStreamScanhistoryFinished(qTaskInfo_t tinfo) {




×







1265



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1266



  return pTaskInfo->streamInfo.recoverScanFinished;




×







1267



}










1268












1269



int32_t qStreamInfoResetTimewindowFilter(qTaskInfo_t tinfo) {




×







1270



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1271



  STimeWindow*   pWindow = &pTaskInfo->streamInfo.fillHistoryWindow;




×







1272












1273



  qDebug("%s remove timeWindow filter:%" PRId64 "-%" PRId64 ", set new window:%" PRId64 "-%" PRId64,





×







1274



         GET_TASKID(pTaskInfo), pWindow->skey, pWindow->ekey, INT64_MIN, INT64_MAX);










1275












1276



  pWindow->skey = INT64_MIN;




×







1277



  pWindow->ekey = INT64_MAX;




×







1278



  return 0;




×







1279



}










1280












1281



void* qExtractReaderFromStreamScanner(void* scanner) {




×







1282



  SStreamScanInfo* pInfo = scanner;




×







1283



  return (void*)pInfo->tqReader;




×







1284



}










1285












1286



const SSchemaWrapper* qExtractSchemaFromTask(qTaskInfo_t tinfo) {




×







1287



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1288



  return pTaskInfo->streamInfo.schema;




×







1289



}










1290












1291



const char* qExtractTbnameFromTask(qTaskInfo_t tinfo) {




×







1292



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1293



  return pTaskInfo->streamInfo.tbName;




×







1294



}










1295












1296



SMqBatchMetaRsp* qStreamExtractMetaMsg(qTaskInfo_t tinfo) {




×







1297



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1298



  return &pTaskInfo->streamInfo.btMetaRsp;




×







1299



}










1300












1301



int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset) {




×







1302



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1303



  tOffsetCopy(pOffset, &pTaskInfo->streamInfo.currentOffset);




×







1304



  return 0;




×







1305



  /*if (code != TSDB_CODE_SUCCESS) {










1306



    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));










1307



    pTaskInfo->code = code;










1308



    T_LONG_JMP(pTaskInfo->env, code);










1309



  }*/










1310



}










1311












1312



int32_t initQueryTableDataCondForTmq(SQueryTableDataCond* pCond, SSnapContext* sContext, SMetaTableInfo* pMtInfo) {




×







1313



  memset(pCond, 0, sizeof(SQueryTableDataCond));




×







1314



  pCond->order = TSDB_ORDER_ASC;




×







1315



  pCond->numOfCols = pMtInfo->schema->nCols;




×







1316



  pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));





×







1317



  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);





×







1318



  if (pCond->colList == NULL || pCond->pSlotList == NULL) {





×







1319



    taosMemoryFreeClear(pCond->colList);





×







1320



    taosMemoryFreeClear(pCond->pSlotList);





×







1321



    return terrno;




×







1322



  }










1323












1324



  pCond->twindows = TSWINDOW_INITIALIZER;




×







1325



  pCond->suid = pMtInfo->suid;




×







1326



  pCond->type = TIMEWINDOW_RANGE_CONTAINED;




×







1327



  pCond->startVersion = -1;




×







1328



  pCond->endVersion = sContext->snapVersion;




×







1329












1330



  for (int32_t i = 0; i < pCond->numOfCols; ++i) {





×







1331



    SColumnInfo* pColInfo = &pCond->colList[i];




×







1332



    pColInfo->type = pMtInfo->schema->pSchema[i].type;




×







1333



    pColInfo->bytes = pMtInfo->schema->pSchema[i].bytes;




×







1334



    pColInfo->colId = pMtInfo->schema->pSchema[i].colId;




×







1335



    pColInfo->pk = pMtInfo->schema->pSchema[i].flags & COL_IS_KEY;




×







1336












1337



    pCond->pSlotList[i] = i;




×







1338



  }










1339












1340



  return TSDB_CODE_SUCCESS;




×







1341



}










1342












1343



void qStreamSetOpen(qTaskInfo_t tinfo) {




×







1344



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1345



  SOperatorInfo* pOperator = pTaskInfo->pRoot;




×







1346



  pOperator->status = OP_NOT_OPENED;




×







1347



}




×







1348












1349



void qStreamSetSourceExcluded(qTaskInfo_t tinfo, int8_t sourceExcluded) {




×







1350



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1351



  pTaskInfo->streamInfo.sourceExcluded = sourceExcluded;




×







1352



}




×







1353












1354



int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType) {




×







1355



  int32_t        code = TSDB_CODE_SUCCESS;




×







1356



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1357



  SStorageAPI*   pAPI = &pTaskInfo->storageAPI;




×







1358












1359



  SOperatorInfo* pOperator = pTaskInfo->pRoot;




×







1360



  const char*    id = GET_TASKID(pTaskInfo);




×







1361












1362



  if (subType == TOPIC_SUB_TYPE__COLUMN && pOffset->type == TMQ_OFFSET__LOG) {





×







1363



    code = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pOperator);




×







1364



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





×







1365



      return code;




×







1366



    }










1367












1368



    SStreamScanInfo* pInfo = pOperator->info;




×







1369



    SStoreTqReader*  pReaderAPI = &pTaskInfo->storageAPI.tqReaderFn;




×







1370



    SWalReader*      pWalReader = pReaderAPI->tqReaderGetWalReader(pInfo->tqReader);




×







1371



    walReaderVerifyOffset(pWalReader, pOffset);




×







1372



  }










1373



  // if pOffset equal to current offset, means continue consume










1374



  if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) {





×







1375



    return 0;




×







1376



  }










1377












1378



  if (subType == TOPIC_SUB_TYPE__COLUMN) {





×







1379



    code = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id, &pOperator);




×







1380



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





×







1381



      return code;




×







1382



    }










1383












1384



    SStreamScanInfo* pInfo = pOperator->info;




×







1385



    STableScanInfo*  pScanInfo = pInfo->pTableScanOp->info;




×







1386



    STableScanBase*  pScanBaseInfo = &pScanInfo->base;




×







1387



    STableListInfo*  pTableListInfo = pScanBaseInfo->pTableListInfo;




×







1388












1389



    if (pOffset->type == TMQ_OFFSET__LOG) {





×







1390



      pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pScanBaseInfo->dataReader);




×







1391



      pScanBaseInfo->dataReader = NULL;




×







1392












1393



      SStoreTqReader* pReaderAPI = &pTaskInfo->storageAPI.tqReaderFn;




×







1394



      SWalReader*     pWalReader = pReaderAPI->tqReaderGetWalReader(pInfo->tqReader);




×







1395



      walReaderVerifyOffset(pWalReader, pOffset);




×







1396



      code = pReaderAPI->tqReaderSeek(pInfo->tqReader, pOffset->version, id);




×







1397



      if (code < 0) {





×







1398



        qError("tqReaderSeek failed ver:%" PRId64 ", %s", pOffset->version, id);





×







1399



        return code;




×







1400



      }










1401



    } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {





×







1402



      // iterate all tables from tableInfoList, and retrieve rows from each table one-by-one










1403



      // those data are from the snapshot in tsdb, besides the data in the wal file.










1404



      int64_t uid = pOffset->uid;




×







1405



      int64_t ts = pOffset->ts;




×







1406



      int32_t index = 0;




×







1407












1408



      // this value may be changed if new tables are created










1409



      taosRLockLatch(&pTaskInfo->lock);




×







1410



      int32_t numOfTables = 0;




×







1411



      code = tableListGetSize(pTableListInfo, &numOfTables);




×







1412



      if (code != TSDB_CODE_SUCCESS) {





×







1413



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1414



        taosRUnLockLatch(&pTaskInfo->lock);




×







1415



        return code;




×







1416



      }










1417












1418



      if (uid == 0) {





×







1419



        if (numOfTables != 0) {





×







1420



          STableKeyInfo* tmp = tableListGetInfo(pTableListInfo, 0);




×







1421



          if (!tmp) {





×







1422



            qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));





×







1423



            taosRUnLockLatch(&pTaskInfo->lock);




×







1424



            return terrno;




×







1425



          }










1426



          if (tmp) uid = tmp->uid;





×







1427



          ts = INT64_MIN;




×







1428



          pScanInfo->currentTable = 0;




×







1429



        } else {










1430



          taosRUnLockLatch(&pTaskInfo->lock);




×







1431



          qError("no table in table list, %s", id);





×







1432



          return TSDB_CODE_TMQ_NO_TABLE_QUALIFIED;




×







1433



        }










1434



      }










1435



      pTaskInfo->storageAPI.tqReaderFn.tqSetTablePrimaryKey(pInfo->tqReader, uid);




×







1436












1437



      qDebug("switch to table uid:%" PRId64 " ts:%" PRId64 "% " PRId64 " rows returned", uid, ts,





×







1438



             pInfo->pTableScanOp->resultInfo.totalRows);










1439



      pInfo->pTableScanOp->resultInfo.totalRows = 0;




×







1440












1441



      // start from current accessed position










1442



      // we cannot start from the pScanInfo->currentTable, since the commit offset may cause the rollback of the start










1443



      // position, let's find it from the beginning.










1444



      index = tableListFind(pTableListInfo, uid, 0);




×







1445



      taosRUnLockLatch(&pTaskInfo->lock);




×







1446












1447



      if (index >= 0) {





×







1448



        pScanInfo->currentTable = index;




×







1449



      } else {










1450



        qError("vgId:%d uid:%" PRIu64 " not found in table list, total:%d, index:%d %s", pTaskInfo->id.vgId, uid,





×







1451



               numOfTables, pScanInfo->currentTable, id);










1452



        return TSDB_CODE_TMQ_NO_TABLE_QUALIFIED;




×







1453



      }










1454












1455



      STableKeyInfo keyInfo = {.uid = uid};




×







1456



      int64_t       oldSkey = pScanBaseInfo->cond.twindows.skey;




×







1457












1458



      // let's start from the next ts that returned to consumer.










1459



      if (pTaskInfo->storageAPI.tqReaderFn.tqGetTablePrimaryKey(pInfo->tqReader)) {





×







1460



        pScanBaseInfo->cond.twindows.skey = ts;




×







1461



      } else {










1462



        pScanBaseInfo->cond.twindows.skey = ts + 1;




×







1463



      }










1464



      pScanInfo->scanTimes = 0;




×







1465












1466



      if (pScanBaseInfo->dataReader == NULL) {





×







1467



        code = pTaskInfo->storageAPI.tsdReader.tsdReaderOpen(pScanBaseInfo->readHandle.vnode, &pScanBaseInfo->cond,




×







1468



                                                             &keyInfo, 1, pScanInfo->pResBlock,










1469



                                                             (void**)&pScanBaseInfo->dataReader, id, NULL);




×







1470



        if (code != TSDB_CODE_SUCCESS) {





×







1471



          qError("prepare read tsdb snapshot failed, uid:%" PRId64 ", code:%s %s", pOffset->uid, tstrerror(code), id);





×







1472



          return code;




×







1473



        }










1474












1475



        qDebug("tsdb reader created with offset(snapshot) uid:%" PRId64 " ts:%" PRId64 " table index:%d, total:%d, %s",





×







1476



               uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id);










1477



      } else {










1478



        code = pTaskInfo->storageAPI.tsdReader.tsdSetQueryTableList(pScanBaseInfo->dataReader, &keyInfo, 1);




×







1479



        if (code != TSDB_CODE_SUCCESS) {





×







1480



          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1481



          return code;




×







1482



        }










1483












1484



        code = pTaskInfo->storageAPI.tsdReader.tsdReaderResetStatus(pScanBaseInfo->dataReader, &pScanBaseInfo->cond);




×







1485



        if (code != TSDB_CODE_SUCCESS) {





×







1486



          qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1487



          return code;




×







1488



        }










1489



        qDebug("tsdb reader offset seek snapshot to uid:%" PRId64 " ts %" PRId64 "  table index:%d numOfTable:%d, %s",





×







1490



               uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id);










1491



      }










1492












1493



      // restore the key value










1494



      pScanBaseInfo->cond.twindows.skey = oldSkey;




×







1495



    } else {










1496



      qError("invalid pOffset->type:%d, %s", pOffset->type, id);





×







1497



      return TSDB_CODE_PAR_INTERNAL_ERROR;




×







1498



    }










1499












1500



  } else {  // subType == TOPIC_SUB_TYPE__TABLE/TOPIC_SUB_TYPE__DB










1501



    if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {





×







1502



      SStreamRawScanInfo* pInfo = pOperator->info;




×







1503



      SSnapContext*       sContext = pInfo->sContext;




×







1504



      SOperatorInfo*      p = NULL;




×







1505












1506



      code = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, id, &p);




×







1507



      if (code != 0) {





×







1508



        return code;




×







1509



      }










1510












1511



      STableListInfo* pTableListInfo = ((SStreamRawScanInfo*)(p->info))->pTableListInfo;




×







1512












1513



      if (pAPI->snapshotFn.setForSnapShot(sContext, pOffset->uid) != 0) {





×







1514



        qError("setDataForSnapShot error. uid:%" PRId64 " , %s", pOffset->uid, id);





×







1515



        return TSDB_CODE_PAR_INTERNAL_ERROR;




×







1516



      }










1517












1518



      SMetaTableInfo mtInfo = {0};




×







1519



      code = pTaskInfo->storageAPI.snapshotFn.getMetaTableInfoFromSnapshot(sContext, &mtInfo);




×







1520



      if (code != 0) {





×







1521



        return code;




×







1522



      }










1523



      pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pInfo->dataReader);




×







1524



      pInfo->dataReader = NULL;




×







1525












1526



      cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);




×







1527



      tableListClear(pTableListInfo);




×







1528












1529



      if (mtInfo.uid == 0) {





×







1530



        tDeleteSchemaWrapper(mtInfo.schema);





×







1531



        goto end;  // no data




×







1532



      }










1533












1534



      pAPI->snapshotFn.taosXSetTablePrimaryKey(sContext, mtInfo.uid);




×







1535



      code = initQueryTableDataCondForTmq(&pTaskInfo->streamInfo.tableCond, sContext, &mtInfo);




×







1536



      if (code != TSDB_CODE_SUCCESS) {





×







1537



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1538



        tDeleteSchemaWrapper(mtInfo.schema);





×







1539



        return code;




×







1540



      }










1541



      if (pAPI->snapshotFn.taosXGetTablePrimaryKey(sContext)) {





×







1542



        pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts;




×







1543



      } else {










1544



        pTaskInfo->streamInfo.tableCond.twindows.skey = pOffset->ts + 1;




×







1545



      }










1546












1547



      code = tableListAddTableInfo(pTableListInfo, mtInfo.uid, 0);




×







1548



      if (code != TSDB_CODE_SUCCESS) {





×







1549



        tDeleteSchemaWrapper(mtInfo.schema);





×







1550



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1551



        return code;




×







1552



      }










1553












1554



      STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0);




×







1555



      if (!pList) {





×







1556



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1557



        tDeleteSchemaWrapper(mtInfo.schema);





×







1558



        return code;




×







1559



      }










1560



      int32_t size = 0;




×







1561



      code = tableListGetSize(pTableListInfo, &size);




×







1562



      if (code != TSDB_CODE_SUCCESS) {





×







1563



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1564



        tDeleteSchemaWrapper(mtInfo.schema);





×







1565



        return code;




×







1566



      }










1567












1568



      code = pTaskInfo->storageAPI.tsdReader.tsdReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size,




×







1569



                                                           NULL, (void**)&pInfo->dataReader, NULL, NULL);




×







1570



      if (code != TSDB_CODE_SUCCESS) {





×







1571



        qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));





×







1572



        tDeleteSchemaWrapper(mtInfo.schema);





×







1573



        return code;




×







1574



      }










1575












1576



      cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);




×







1577



      tstrncpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName, TSDB_TABLE_NAME_LEN);




×







1578



//      pTaskInfo->streamInfo.suid = mtInfo.suid == 0 ? mtInfo.uid : mtInfo.suid;










1579



      tDeleteSchemaWrapper(pTaskInfo->streamInfo.schema);





×







1580



      pTaskInfo->streamInfo.schema = mtInfo.schema;




×







1581












1582



      qDebug("tmqsnap qStreamPrepareScan snapshot data uid:%" PRId64 " ts %" PRId64 " %s", mtInfo.uid, pOffset->ts, id);





×







1583



    } else if (pOffset->type == TMQ_OFFSET__SNAPSHOT_META) {





×







1584



      SStreamRawScanInfo* pInfo = pOperator->info;




×







1585



      SSnapContext*       sContext = pInfo->sContext;




×







1586



      code = pTaskInfo->storageAPI.snapshotFn.setForSnapShot(sContext, pOffset->uid);




×







1587



      if (code != 0) {





×







1588



        qError("setForSnapShot error. uid:%" PRIu64 " ,version:%" PRId64, pOffset->uid, pOffset->version);





×







1589



        return code;




×







1590



      }










1591



      qDebug("tmqsnap qStreamPrepareScan snapshot meta uid:%" PRId64 " ts %" PRId64 " %s", pOffset->uid, pOffset->ts,





×







1592



             id);










1593



    } else if (pOffset->type == TMQ_OFFSET__LOG) {





×







1594



      SStreamRawScanInfo* pInfo = pOperator->info;




×







1595



      pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pInfo->dataReader);




×







1596



      pInfo->dataReader = NULL;




×







1597



      qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id);





×







1598



    }










1599



  }










1600












1601



end:




×







1602



  tOffsetCopy(&pTaskInfo->streamInfo.currentOffset, pOffset);




×







1603



  return 0;




×







1604



}










1605












1606



void qProcessRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {




×







1607



  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;




×







1608



  if (pMsg->info.ahandle == NULL) {





×







1609



    qError("pMsg->info.ahandle is NULL");





×







1610



    return;




×







1611



  }










1612












1613



  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL};




×







1614












1615



  if (pMsg->contLen > 0) {





×







1616



    buf.pData = taosMemoryCalloc(1, pMsg->contLen);





×







1617



    if (buf.pData == NULL) {





×







1618



      pMsg->code = terrno;




×







1619



    } else {










1620



      memcpy(buf.pData, pMsg->pCont, pMsg->contLen);




×







1621



    }










1622



  }










1623












1624



  (void)pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);




×







1625



  rpcFreeCont(pMsg->pCont);




×







1626



  destroySendMsgInfo(pSendInfo);




×







1627



}










1628












1629



SArray* qGetQueriedTableListInfo(qTaskInfo_t tinfo) {




×







1630



  int32_t        code = TSDB_CODE_SUCCESS;




×







1631



  int32_t        lino = 0;




×







1632



  SExecTaskInfo* pTaskInfo = tinfo;




×







1633



  SArray*        plist = NULL;




×







1634












1635



  code = getTableListInfo(pTaskInfo, &plist);




×







1636



  if (code || plist == NULL) {





×







1637



    return NULL;




×







1638



  }










1639












1640



  // only extract table in the first elements










1641



  STableListInfo* pTableListInfo = taosArrayGetP(plist, 0);




×







1642












1643



  SArray* pUidList = taosArrayInit(10, sizeof(uint64_t));




×







1644



  QUERY_CHECK_NULL(pUidList, code, lino, _end, terrno);





×







1645












1646



  int32_t numOfTables = 0;




×







1647



  code = tableListGetSize(pTableListInfo, &numOfTables);




×







1648



  QUERY_CHECK_CODE(code, lino, _end);





×







1649












1650



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





×







1651



    STableKeyInfo* pKeyInfo = tableListGetInfo(pTableListInfo, i);




×







1652



    QUERY_CHECK_NULL(pKeyInfo, code, lino, _end, terrno);





×







1653



    void* tmp = taosArrayPush(pUidList, &pKeyInfo->uid);




×







1654



    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







1655



  }










1656












1657



  taosArrayDestroy(plist);




×







1658












1659



_end:




×







1660



  if (code != TSDB_CODE_SUCCESS) {





×







1661



    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));





×







1662



    T_LONG_JMP(pTaskInfo->env, code);





×







1663



  }










1664



  return pUidList;




×







1665



}










1666












1667



static int32_t extractTableList(SArray* pList, const SOperatorInfo* pOperator) {




×







1668



  int32_t        code = TSDB_CODE_SUCCESS;




×







1669



  int32_t        lino = 0;




×







1670



  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;




×







1671












1672



  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {





×







1673



    SStreamScanInfo* pScanInfo = pOperator->info;




×







1674



    STableScanInfo*  pTableScanInfo = pScanInfo->pTableScanOp->info;




×







1675












1676



    void* tmp = taosArrayPush(pList, &pTableScanInfo->base.pTableListInfo);




×







1677



    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







1678



  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {





×







1679



    STableScanInfo* pScanInfo = pOperator->info;




×







1680












1681



    void* tmp = taosArrayPush(pList, &pScanInfo->base.pTableListInfo);




×







1682



    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);





×







1683



  } else {










1684



    if (pOperator->pDownstream != NULL && pOperator->pDownstream[0] != NULL) {





×







1685



      code = extractTableList(pList, pOperator->pDownstream[0]);




×







1686



    }










1687



  }










1688












1689



_end:




×







1690



  if (code != TSDB_CODE_SUCCESS) {





×







1691



    qError("%s %s failed at line %d since %s", pTaskInfo->id.str, __func__, lino, tstrerror(code));





×







1692



  }










1693



  return code;




×







1694



}










1695












1696



int32_t getTableListInfo(const SExecTaskInfo* pTaskInfo, SArray** pList) {




×







1697



  if (pList == NULL) {





×







1698



    return TSDB_CODE_INVALID_PARA;




×







1699



  }










1700












1701



  *pList = NULL;




×







1702



  SArray* pArray = taosArrayInit(0, POINTER_BYTES);




×







1703



  if (pArray == NULL) {





×







1704



    return terrno;




×







1705



  }










1706












1707



  int32_t code = extractTableList(pArray, pTaskInfo->pRoot);




×







1708



  if (code == 0) {





×







1709



    *pList = pArray;




×







1710



  } else {










1711



    taosArrayDestroy(pArray);




×







1712



  }










1713



  return code;




×







1714



}










1715












1716



int32_t qStreamOperatorReleaseState(qTaskInfo_t tInfo) {




×







1717



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tInfo;




×







1718



  if (pTaskInfo->pRoot->fpSet.releaseStreamStateFn != NULL) {





×







1719



    pTaskInfo->pRoot->fpSet.releaseStreamStateFn(pTaskInfo->pRoot);




×







1720



  }










1721



  return 0;




×







1722



}










1723












1724



int32_t qStreamOperatorReloadState(qTaskInfo_t tInfo) {




×







1725



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tInfo;




×







1726



  if (pTaskInfo->pRoot->fpSet.reloadStreamStateFn != NULL) {





×







1727



    pTaskInfo->pRoot->fpSet.reloadStreamStateFn(pTaskInfo->pRoot);




×







1728



  }










1729



  return 0;




×







1730



}










1731












1732



void qResetTaskCode(qTaskInfo_t tinfo) {




×







1733



  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;




×







1734












1735



  int32_t code = pTaskInfo->code;




×







1736



  pTaskInfo->code = 0;




×







1737



  qDebug("0x%" PRIx64 " reset task code to be success, prev:%s", pTaskInfo->id.taskId, tstrerror(code));





×







1738



}




×





















    

  • 

    •