taosdata / TDengine / #4849

coverage: 63.829% (+0.6%) from 63.27%
#4849

push

travis-ci

guanshengliang 
test: enable auto upload

148928 of 233322 relevant lines covered (63.83%)

115593255.19 hits per line

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

88.94
/source/dnode/vnode/src/tsdb/tsdbMergeTree.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 "tsdb.h"










17



#include "tsdbFSet2.h"










18



#include "tsdbMerge.h"










19



#include "tsdbReadUtil.h"










20



#include "tsdbSttFileRW.h"










21



#include "tsdbUtil2.h"










22












23



static void tLDataIterClose2(SLDataIter *pIter);










24












25



// SLDataIter =================================================










26



int32_t tCreateSttBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols, SSttBlockLoadInfo **pInfo) {




170,660,848✔





27



  *pInfo = NULL;




170,660,848✔





28












29



  SSttBlockLoadInfo *pLoadInfo = taosMemoryCalloc(1, sizeof(SSttBlockLoadInfo));




170,706,268✔





30



  if (pLoadInfo == NULL) {




170,603,486✔





31



    return terrno;




×







32



  }










33












34



  pLoadInfo->blockData[0].sttBlockIndex = -1;




170,603,486✔





35



  pLoadInfo->blockData[1].sttBlockIndex = -1;




170,611,005✔





36












37



  pLoadInfo->currentLoadBlockIndex = 1;




170,617,973✔





38












39



  int32_t code = tBlockDataCreate(&pLoadInfo->blockData[0].data);




170,644,213✔





40



  if (code) {




170,662,949✔





41



    taosMemoryFreeClear(pLoadInfo);




×







42



    return code;




×







43



  }










44












45



  code = tBlockDataCreate(&pLoadInfo->blockData[1].data);




170,662,949✔





46



  if (code) {




170,654,858✔





47



    taosMemoryFreeClear(pLoadInfo);




×







48



    return code;




×







49



  }










50












51



  pLoadInfo->aSttBlk = taosArrayInit(4, sizeof(SSttBlk));




170,654,858✔





52



  if (pLoadInfo->aSttBlk == NULL) {




170,717,331✔





53



    taosMemoryFreeClear(pLoadInfo);




×







54



    return terrno;




×







55



  }










56












57



  pLoadInfo->pSchema = pSchema;




170,726,455✔





58



  pLoadInfo->colIds = colList;




170,743,508✔





59



  pLoadInfo->numOfCols = numOfCols;




170,741,577✔





60












61



  *pInfo = pLoadInfo;




170,726,488✔





62



  return code;




170,712,695✔





63



}










64












65



static void freeItem(void *pValue) {




2,147,483,647✔





66



  SValue *p = (SValue *)pValue;




2,147,483,647✔





67



  if (IS_VAR_DATA_TYPE(p->type) || p->type == TSDB_DATA_TYPE_DECIMAL) {




2,147,483,647✔





68



    taosMemoryFree(p->pData);




5,670,550✔





69



  }










70



}




2,147,483,647✔





71












72



void destroySttBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo) {




170,656,524✔





73



  if (pLoadInfo == NULL) {




170,656,524✔





74



    return;




×







75



  }










76












77



  pLoadInfo->currentLoadBlockIndex = 1;




170,656,524✔





78












79



  SBlockDataInfo *pInfo = &pLoadInfo->blockData[0];




170,713,205✔





80



  tBlockDataDestroy(&pInfo->data);




170,668,301✔





81



  pInfo->sttBlockIndex = -1;




170,692,897✔





82



  pInfo->pin = false;




170,684,901✔





83












84



  pInfo = &pLoadInfo->blockData[1];




170,688,979✔





85



  tBlockDataDestroy(&pInfo->data);




170,705,212✔





86



  pInfo->sttBlockIndex = -1;




170,716,872✔





87



  pInfo->pin = false;




170,704,782✔





88












89



  taosArrayDestroy(pLoadInfo->info.pUid);




170,745,959✔





90



  taosArrayDestroyEx(pLoadInfo->info.pFirstKey, freeItem);




170,704,433✔





91



  taosArrayDestroyEx(pLoadInfo->info.pLastKey, freeItem);




170,678,833✔





92



  taosArrayDestroy(pLoadInfo->info.pCount);




170,722,480✔





93



  taosArrayDestroy(pLoadInfo->info.pFirstTs);




170,717,250✔





94



  taosArrayDestroy(pLoadInfo->info.pLastTs);




170,708,565✔





95












96



  pLoadInfo->info.pUid = NULL;




170,703,905✔





97



  pLoadInfo->info.pFirstKey = NULL;




170,704,828✔





98



  pLoadInfo->info.pLastKey = NULL;




170,709,288✔





99



  pLoadInfo->info.pCount = NULL;




170,737,850✔





100



  pLoadInfo->info.pFirstTs = NULL;




170,726,069✔





101



  pLoadInfo->info.pLastTs = NULL;




170,706,177✔





102












103



  taosArrayDestroy(pLoadInfo->aSttBlk);




170,705,846✔





104



  taosMemoryFree(pLoadInfo);




170,707,956✔





105



}










106












107



void destroyLDataIter(SLDataIter *pIter) {




170,740,133✔





108



  tLDataIterClose2(pIter);




170,740,133✔





109



  destroySttBlockLoadInfo(pIter->pBlockLoadInfo);




170,685,076✔





110



  taosMemoryFree(pIter);




170,705,384✔





111



}




170,685,627✔





112












113



void destroySttBlockReader(SArray *pLDataIterArray, SSttBlockLoadCostInfo *pLoadCost) {




307,139,955✔





114



  if (pLDataIterArray == NULL) {




307,139,955✔





115



    return;




3,566,394✔





116



  }










117












118



  int32_t numOfLevel = taosArrayGetSize(pLDataIterArray);




303,573,561✔





119



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




474,187,703✔





120



    SArray *pList = taosArrayGetP(pLDataIterArray, i);




170,583,730✔





121



    for (int32_t j = 0; j < taosArrayGetSize(pList); ++j) {




341,307,375✔





122



      SLDataIter *pIter = taosArrayGetP(pList, j);




170,750,047✔





123



      if (pIter->pBlockLoadInfo != NULL) {




170,750,658✔





124



        SSttBlockLoadCostInfo *pCost = &pIter->pBlockLoadInfo->cost;




170,760,276✔





125



        if (pLoadCost != NULL) {




170,750,967✔





126



          pLoadCost->loadBlocks += pCost->loadBlocks;




170,748,753✔





127



          pLoadCost->loadStatisBlocks += pCost->loadStatisBlocks;




170,759,479✔





128



          pLoadCost->blockElapsedTime += pCost->blockElapsedTime;




170,758,963✔





129



          pLoadCost->statisElapsedTime += pCost->statisElapsedTime;




170,739,883✔





130



        }










131



      }










132












133



      destroyLDataIter(pIter);




170,747,040✔





134



    }










135












136



    taosArrayDestroy(pList);




170,562,994✔





137



  }










138












139



  taosArrayDestroy(pLDataIterArray);




303,603,973✔





140



}










141












142



// choose the unpinned slot to load next data block










143



static void updateBlockLoadSlot(SSttBlockLoadInfo *pLoadInfo) {




112,824,342✔





144



  int32_t nextSlotIndex = pLoadInfo->currentLoadBlockIndex ^ 1;




112,824,342✔





145



  if (pLoadInfo->blockData[nextSlotIndex].pin) {




112,891,716✔





146



    nextSlotIndex = nextSlotIndex ^ 1;




×







147



  }










148












149



  pLoadInfo->currentLoadBlockIndex = nextSlotIndex;




112,861,762✔





150



}




112,865,695✔





151












152



static int32_t loadLastBlock(SLDataIter *pIter, const char *idStr, SBlockData **pResBlock) {




2,147,483,647✔





153



  if (pResBlock != NULL) {




2,147,483,647✔





154



    *pResBlock = NULL;




2,147,483,647✔





155



  }










156












157



  int32_t            code = 0;




2,147,483,647✔





158



  SSttBlockLoadInfo *pInfo = pIter->pBlockLoadInfo;




2,147,483,647✔





159












160



  if (pInfo->blockData[0].sttBlockIndex == pIter->iSttBlk) {




2,147,483,647✔





161



    if (pInfo->currentLoadBlockIndex != 0) {




2,147,483,647✔





162



      tsdbDebug("current load index is set to 0, block index:%d, fileVer:%" PRId64 ", due to uid:%" PRIu64




12,057✔





163



                ", load data, %s",










164



                pIter->iSttBlk, pIter->cid, pIter->uid, idStr);










165



      pInfo->currentLoadBlockIndex = 0;




12,057✔





166



    }










167












168



    *pResBlock = &pInfo->blockData[0].data;




2,147,483,647✔





169



    return code;




2,147,483,647✔





170



  }










171












172



  if (pInfo->blockData[1].sttBlockIndex == pIter->iSttBlk) {




2,147,483,647✔





173



    if (pInfo->currentLoadBlockIndex != 1) {




2,147,483,647✔





174



      tsdbDebug("current load index is set to 1, block index:%d, fileVer:%" PRId64 ", due to uid:%" PRIu64




5,877✔





175



                ", load data, %s",










176



                pIter->iSttBlk, pIter->cid, pIter->uid, idStr);










177



      pInfo->currentLoadBlockIndex = 1;




5,877✔





178



    }










179












180



    *pResBlock = &pInfo->blockData[1].data;




2,147,483,647✔





181



    return code;




2,147,483,647✔





182



  }










183












184



  if (pIter->pSttBlk == NULL || pInfo->pSchema == NULL) {




112,813,928✔





185



    return code;




×







186



  }










187












188



  updateBlockLoadSlot(pInfo);




112,855,607✔





189



  int64_t st = taosGetTimestampUs();




112,895,571✔





190












191



  SBlockData *pBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex].data;




112,895,571✔





192



  code = tsdbSttFileReadBlockDataByColumn(pIter->pReader, pIter->pSttBlk, pBlock, pInfo->pSchema, &pInfo->colIds[1],




112,826,313✔





193



                                          pInfo->numOfCols - 1);




112,847,456✔





194



  if (code != TSDB_CODE_SUCCESS) {




112,853,813✔





195



    return code;




×







196



  }










197












198



  double el = (taosGetTimestampUs() - st) / 1000.0;




112,871,313✔





199



  pInfo->cost.blockElapsedTime += el;




112,871,313✔





200



  pInfo->cost.loadBlocks += 1;




112,864,895✔





201












202



  tsdbDebug("read stt block, total load:%" PRId64 ", trigger by uid:%" PRIu64 ", stt-fileVer:%" PRId64




112,855,059✔





203



            ", last block index:%d, entry:%d, rows:%d, uidRange:%" PRId64 "-%" PRId64 " tsRange:%" PRId64 "-%" PRId64










204



            " %p, elapsed time:%.2f ms, %s",










205



            pInfo->cost.loadBlocks, pIter->uid, pIter->cid, pIter->iSttBlk, pInfo->currentLoadBlockIndex, pBlock->nRow,










206



            pIter->pSttBlk->minUid, pIter->pSttBlk->maxUid, pIter->pSttBlk->minKey, pIter->pSttBlk->maxKey, pBlock, el,










207



            idStr);










208












209



  pInfo->blockData[pInfo->currentLoadBlockIndex].sttBlockIndex = pIter->iSttBlk;




112,855,059✔





210



  pIter->iRow = (pIter->backward) ? pInfo->blockData[pInfo->currentLoadBlockIndex].data.nRow : -1;




112,872,826✔





211












212



  tsdbDebug("last block index list:%d, %d, rowIndex:%d %s", pInfo->blockData[0].sttBlockIndex,




112,877,453✔





213



            pInfo->blockData[1].sttBlockIndex, pIter->iRow, idStr);










214












215



  *pResBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex].data;




112,878,205✔





216



  return code;




112,826,328✔





217



}










218












219



// find the earliest block that contains the required records










220



static FORCE_INLINE int32_t findEarliestIndex(int32_t index, uint64_t uid, const SSttBlk *pBlockList, int32_t num,










221



                                              int32_t backward) {










222



  int32_t i = index;




127,762,153✔





223



  int32_t step = backward ? 1 : -1;




127,762,153✔





224



  while (i >= 0 && i < num && uid >= pBlockList[i].minUid && uid <= pBlockList[i].maxUid) {




256,729,558✔





225



    i += step;




128,967,405✔





226



  }










227



  return i - step;




127,723,507✔





228



}










229












230



static int32_t binarySearchForStartBlock(SSttBlk *pBlockList, int32_t num, uint64_t uid, int32_t backward) {




236,512,488✔





231



  int32_t midPos = -1;




236,512,488✔





232



  if (num <= 0) {




236,512,488✔





233



    return -1;




48,926,100✔





234



  }










235












236



  int32_t firstPos = 0;




187,586,388✔





237



  int32_t lastPos = num - 1;




187,586,388✔





238












239



  // find the first position which is bigger than the key










240



  if ((uid > pBlockList[lastPos].maxUid) || (uid < pBlockList[firstPos].minUid)) {




187,586,388✔





241



    return -1;




59,875,131✔





242



  }










243












244



  while (1) {




5,939,727✔





245



    if (uid >= pBlockList[firstPos].minUid && uid <= pBlockList[firstPos].maxUid) {




133,663,904✔





246



      return findEarliestIndex(firstPos, uid, pBlockList, num, backward);




123,207,797✔





247



    }










248












249



    if (uid > pBlockList[lastPos].maxUid || uid < pBlockList[firstPos].minUid) {




10,470,096✔





250



      return -1;




8,722✔





251



    }










252












253



    int32_t numOfRows = lastPos - firstPos + 1;




10,455,804✔





254



    midPos = (numOfRows >> 1u) + firstPos;




10,455,804✔





255












256



    if (uid < pBlockList[midPos].minUid) {




10,455,804✔





257



      lastPos = midPos - 1;




2,720,150✔





258



    } else if (uid > pBlockList[midPos].maxUid) {




7,735,421✔





259



      firstPos = midPos + 1;




3,219,577✔





260



    } else {










261



      return findEarliestIndex(midPos, uid, pBlockList, num, backward);




4,515,710✔





262



    }










263



  }










264



}










265












266



static FORCE_INLINE int32_t findEarliestRow(int32_t index, uint64_t uid, const uint64_t *uidList, int32_t num,










267



                                            int32_t backward) {










268



  int32_t i = index;




128,039,757✔





269



  int32_t step = backward ? 1 : -1;




128,039,757✔





270



  while (i >= 0 && i < num && uid == uidList[i]) {




2,147,483,647✔





271



    i += step;




2,111,078,267✔





272



  }










273



  return i - step;




128,051,176✔





274



}










275












276



static int32_t binarySearchForStartRowIndex(uint64_t *uidList, int32_t num, uint64_t uid, int32_t backward) {




128,098,305✔





277



  int32_t firstPos = 0;




128,098,305✔





278



  int32_t lastPos = num - 1;




128,098,305✔





279












280



  // find the first position which is bigger than the key










281



  if ((uid > uidList[lastPos]) || (uid < uidList[firstPos])) {




128,098,305✔





282



    return -1;




×







283



  }










284












285



  while (1) {




33,756,707✔





286



    if (uid == uidList[firstPos]) {




161,889,081✔





287



      return findEarliestRow(firstPos, uid, uidList, num, backward);




102,885,321✔





288



    }










289












290



    if (uid > uidList[lastPos] || uid < uidList[firstPos]) {




58,989,011✔





291



      return -1;




67,571✔





292



    }










293












294



    int32_t numOfRows = lastPos - firstPos + 1;




58,922,439✔





295



    int32_t midPos = (numOfRows >> 1u) + firstPos;




58,922,439✔





296












297



    if (uid < uidList[midPos]) {




58,922,439✔





298



      lastPos = midPos - 1;




10,895,952✔





299



    } else if (uid > uidList[midPos]) {




48,026,417✔





300



      firstPos = midPos + 1;




22,860,755✔





301



    } else {










302



      return findEarliestRow(midPos, uid, uidList, num, backward);




25,165,855✔





303



    }










304



  }










305



}










306












307



static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray, SSttBlockLoadInfo *pBlockLoadInfo,




170,659,905✔





308



                                   uint64_t suid) {










309



  void   *px = NULL;




170,659,905✔





310



  int32_t code = TSDB_CODE_SUCCESS;




170,659,905✔





311



  if (TARRAY2_SIZE(pArray) <= 0) {




170,659,905✔





312



    return code;




145,742✔





313



  }










314












315



  SSttBlk *pStart = &pArray->data[0];




170,547,069✔





316



  SSttBlk *pEnd = &pArray->data[TARRAY2_SIZE(pArray) - 1];




170,538,636✔





317












318



  // all identical










319



  if (pStart->suid == pEnd->suid) {




170,516,022✔





320



    if (pStart->suid != suid) {  // no qualified stt block existed




145,461,310✔





321



      taosArrayClear(pBlockLoadInfo->aSttBlk);




32,750,803✔





322



      pIter->iSttBlk = -1;




32,745,066✔





323



      return TSDB_CODE_SUCCESS;




32,749,429✔





324



    } else {  // all blocks are qualified










325



      taosArrayClear(pBlockLoadInfo->aSttBlk);




112,704,130✔





326



      px = taosArrayAddBatch(pBlockLoadInfo->aSttBlk, pArray->data, pArray->size);




112,730,909✔





327



      if (px == NULL) {




112,742,881✔





328



        return terrno;




×







329



      }










330



    }










331



  } else {










332



    SArray *pTmp = taosArrayInit(TARRAY2_SIZE(pArray), sizeof(SSttBlk));




25,064,695✔





333



    if (pTmp == NULL) {




25,069,048✔





334



      return terrno;




×







335



    }










336












337



    for (int32_t i = 0; i < TARRAY2_SIZE(pArray); ++i) {




84,287,106✔





338



      SSttBlk *p = &pArray->data[i];




68,895,385✔





339



      if (p->suid < suid) {




68,893,836✔





340



        continue;




38,187,269✔





341



      }










342












343



      if (p->suid == suid) {




30,712,878✔





344



        void *px = taosArrayPush(pTmp, p);




21,030,549✔





345



        if (px == NULL) {




21,030,549✔





346



          code = terrno;




×







347



          break;




×







348



        }










349



      } else if (p->suid > suid) {




9,682,052✔





350



        break;




9,681,821✔





351



      }










352



    }










353












354



    taosArrayDestroy(pBlockLoadInfo->aSttBlk);




25,066,723✔





355



    pBlockLoadInfo->aSttBlk = pTmp;




25,067,924✔





356



  }










357












358



  return code;




137,769,784✔





359



}










360












361



static int32_t tValueDupPayload(SValue *pVal) {




20,352,492✔





362



  if (IS_VAR_DATA_TYPE(pVal->type) || pVal->type == TSDB_DATA_TYPE_DECIMAL) {




20,352,492✔





363



    char *p = (char *)pVal->pData;




5,667,516✔





364



    char *pBuf = taosMemoryMalloc(pVal->nData);




5,669,264✔





365



    if (pBuf == NULL) {




5,669,445✔





366



      return terrno;




×







367



    }










368












369



    memcpy(pBuf, p, pVal->nData);




5,669,445✔





370



    pVal->pData = (uint8_t *)pBuf;




5,669,599✔





371



  }










372












373



  return TSDB_CODE_SUCCESS;




20,354,421✔





374



}










375












376



static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBlockLoadInfo *pBlockLoadInfo,




170,670,137✔





377



                                          TStatisBlkArray *pStatisBlkArray, uint64_t suid, const char *id) {










378



  int32_t code = TSDB_CODE_SUCCESS;




170,670,137✔





379



  int32_t lino = 0;




170,670,137✔





380



  void   *px = NULL;




170,670,137✔





381



  int32_t startIndex = 0;




170,670,137✔





382



  int32_t ret = 0;




170,670,137✔





383












384



  int32_t numOfBlocks = TARRAY2_SIZE(pStatisBlkArray);




170,670,137✔





385



  if (numOfBlocks <= 0) {




170,647,219✔





386



    return code;




145,742✔





387



  }










388












389



  while ((startIndex < numOfBlocks) && (pStatisBlkArray->data[startIndex].maxTbid.suid < suid)) {




186,949,114✔





390



    ++startIndex;




16,447,637✔





391



  }










392












393



  if (startIndex >= numOfBlocks || pStatisBlkArray->data[startIndex].minTbid.suid > suid) {




170,519,970✔





394



    return 0;




39,048,027✔





395



  }










396












397



  int32_t endIndex = startIndex;




131,477,902✔





398



  while (endIndex < numOfBlocks && pStatisBlkArray->data[endIndex].minTbid.suid <= suid) {




265,489,715✔





399



    ++endIndex;




134,011,813✔





400



  }










401












402



  int32_t num = endIndex - startIndex;




131,434,293✔





403



  pBlockLoadInfo->cost.loadStatisBlocks += num;




131,434,293✔





404












405



  STbStatisBlock block;




131,478,919✔





406



  code = tStatisBlockInit(&block);




131,522,275✔





407



  QUERY_CHECK_CODE(code, lino, _end);




131,539,871✔





408












409



  int64_t st = taosGetTimestampUs();




131,537,635✔





410












411



  for (int32_t k = startIndex; k < endIndex; ++k) {




265,564,458✔





412



    code = tsdbSttFileReadStatisBlock(pSttFileReader, &pStatisBlkArray->data[k], &block);




133,991,344✔





413



    QUERY_CHECK_CODE(code, lino, _end);




134,054,505✔





414












415



    int32_t i = 0;




134,054,505✔





416



    int32_t rows = block.numOfRecords;




134,054,505✔





417



    while (i < rows && ((int64_t *)block.suids.data)[i] != suid) {




156,598,255✔





418



      ++i;




22,543,750✔





419



    }










420












421



    // existed










422



    if (i < rows) {




134,073,504✔





423



      SSttTableRowsInfo *pInfo = &pBlockLoadInfo->info;




133,791,084✔





424












425



      if (pInfo->pUid == NULL) {




133,789,068✔





426



        pInfo->pUid = taosArrayInit(rows, sizeof(int64_t));




131,274,961✔





427



        pInfo->pFirstTs = taosArrayInit(rows, sizeof(int64_t));




131,274,903✔





428



        pInfo->pLastTs = taosArrayInit(rows, sizeof(int64_t));




131,274,529✔





429



        pInfo->pCount = taosArrayInit(rows, sizeof(int64_t));




131,278,502✔





430












431



        pInfo->pFirstKey = taosArrayInit(rows, sizeof(SValue));




131,279,914✔





432



        pInfo->pLastKey = taosArrayInit(rows, sizeof(SValue));




131,247,349✔





433












434



        if (pInfo->pUid == NULL || pInfo->pFirstTs == NULL || pInfo->pLastTs == NULL || pInfo->pCount == NULL ||




131,242,296✔





435



            pInfo->pFirstKey == NULL || pInfo->pLastKey == NULL) {




131,279,428✔





436



          code = terrno;




12,530✔





437



          goto _end;




×







438



        }










439



      }










440












441



      if (pStatisBlkArray->data[k].maxTbid.suid == suid) {




133,773,773✔





442



        int32_t size = rows - i;




126,256,125✔





443



        int32_t offset = i * sizeof(int64_t);




126,256,125✔





444












445



        px = taosArrayAddBatch(pInfo->pUid, tBufferGetDataAt(&block.uids, offset), size);




126,256,125✔





446



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




126,263,124✔





447












448



        px = taosArrayAddBatch(pInfo->pFirstTs, tBufferGetDataAt(&block.firstKeyTimestamps, offset), size);




126,263,124✔





449



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




126,246,304✔





450












451



        px = taosArrayAddBatch(pInfo->pLastTs, tBufferGetDataAt(&block.lastKeyTimestamps, offset), size);




126,246,304✔





452



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




126,254,861✔





453












454



        px = taosArrayAddBatch(pInfo->pCount, tBufferGetDataAt(&block.counts, offset), size);




126,254,861✔





455



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




126,189,501✔





456












457



        if (block.numOfPKs > 0) {




126,189,501✔





458



          SValue vFirst = {0}, vLast = {0};




5,813,726✔





459



          for (int32_t f = i; f < rows; ++f) {




12,749,329✔





460



            code = tValueColumnGet(&block.firstKeyPKs[0], f, &vFirst);




6,933,031✔





461



            TSDB_CHECK_CODE(code, lino, _end);




6,931,745✔





462












463



            code = tValueDupPayload(&vFirst);




6,931,745✔





464



            TSDB_CHECK_CODE(code, lino, _end);




6,933,520✔





465












466



            px = taosArrayPush(pInfo->pFirstKey, &vFirst);




6,933,520✔





467



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




6,934,471✔





468












469



            // todo add api to clone the original data










470



            code = tValueColumnGet(&block.lastKeyPKs[0], f, &vLast);




6,934,471✔





471



            TSDB_CHECK_CODE(code, lino, _end);




6,933,185✔





472












473



            code = tValueDupPayload(&vLast);




6,933,185✔





474



            TSDB_CHECK_CODE(code, lino, _end);




6,934,317✔





475












476



            px = taosArrayPush(pInfo->pLastKey, &vLast);




6,934,317✔





477



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




6,934,471✔





478



          }










479



        } else {










480



          SValue vFirst = {0};




120,375,775✔





481



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




1,835,084,073✔





482



            px = taosArrayPush(pInfo->pFirstKey, &vFirst);




1,715,284,516✔





483



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




1,715,157,605✔





484












485



            px = taosArrayPush(pInfo->pLastKey, &vFirst);




1,715,157,605✔





486



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




1,714,682,498✔





487



          }










488



        }










489



      } else {










490



        STbStatisRecord record = {0};




7,548,083✔





491



        while (i < rows) {




23,018,615✔





492



          code = tStatisBlockGet(&block, i, &record);




23,018,003✔





493



          TSDB_CHECK_CODE(code, lino, _end);




23,017,775✔





494












495



          if (record.suid != suid) {




23,017,775✔





496



            break;




7,549,891✔





497



          }










498












499



          px = taosArrayPush(pInfo->pUid, &record.uid);




15,467,884✔





500



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




15,470,048✔





501












502



          px = taosArrayPush(pInfo->pCount, &record.count);




15,470,048✔





503



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




15,472,460✔





504












505



          px = taosArrayPush(pInfo->pFirstTs, &record.firstKey.ts);




15,472,460✔





506



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




15,472,220✔





507












508



          px = taosArrayPush(pInfo->pLastTs, &record.lastKey.ts);




15,472,220✔





509



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




15,472,460✔





510












511



          if (record.firstKey.numOfPKs > 0) {




15,472,460✔





512



            SValue s = record.firstKey.pks[0];




3,243,935✔





513



            code = tValueDupPayload(&s);




3,243,935✔





514



            TSDB_CHECK_CODE(code, lino, _end);




3,243,935✔





515












516



            px = taosArrayPush(pInfo->pFirstKey, &s);




3,243,935✔





517



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




3,243,935✔





518












519



            s = record.lastKey.pks[0];




3,243,935✔





520



            code = tValueDupPayload(&s);




3,243,935✔





521



            TSDB_CHECK_CODE(code, lino, _end);




3,243,935✔





522












523



            px = taosArrayPush(pInfo->pLastKey, &s);




3,243,935✔





524



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




3,243,935✔





525



          } else {










526



            SValue v = {0};




12,228,525✔





527



            px = taosArrayPush(pInfo->pFirstKey, &v);




12,226,229✔





528



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




12,228,165✔





529












530



            px = taosArrayPush(pInfo->pLastKey, &v);




12,228,165✔





531



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




12,227,805✔





532



          }










533












534



          i += 1;




15,469,692✔





535



        }










536



      }










537



    }










538



  }










539












540



_end:




131,640,068✔





541



  tStatisBlockDestroy(&block);




131,508,313✔





542



  if (code != 0) {




131,538,070✔





543



    tsdbError("%s error happens at:%s line number: %d, code:%s", id, __func__, lino, tstrerror(code));




×







544



  } else {










545



    double el = (taosGetTimestampUs() - st) / 1000.0;




131,538,947✔





546



    pBlockLoadInfo->cost.statisElapsedTime += el;




131,538,947✔





547












548



    tsdbDebug("%s load %d statis blocks into buf, elapsed time:%.2fms", id, num, el);




131,517,981✔





549



  }










550



  return code;




131,494,703✔





551



}










552












553



static int32_t doLoadSttFilesBlk(SSttBlockLoadInfo *pBlockLoadInfo, SLDataIter *pIter, int64_t suid,




170,658,634✔





554



                                 _load_tomb_fn loadTombFn, void *pReader1, const char *idStr) {










555



  int64_t st = taosGetTimestampUs();




170,725,470✔





556












557



  const TSttBlkArray *pSttBlkArray = NULL;




170,725,470✔





558



  pBlockLoadInfo->sttBlockLoaded = true;




170,713,666✔





559












560



  // load the stt block info for each stt-block










561



  int32_t code = tsdbSttFileReadSttBlk(pIter->pReader, &pSttBlkArray);




170,695,818✔





562



  if (code != TSDB_CODE_SUCCESS) {




170,642,735✔





563



    tsdbError("load stt blk failed, code:%s, %s", tstrerror(code), idStr);




×







564



    return code;




×







565



  }










566












567



  // load the stt block info for each stt file block










568



  code = extractSttBlockInfo(pIter, pSttBlkArray, pBlockLoadInfo, suid);




170,642,735✔





569



  if (code != TSDB_CODE_SUCCESS) {




170,654,061✔





570



    tsdbError("load stt block info failed, code:%s, %s", tstrerror(code), idStr);




×







571



    return code;




×







572



  }










573












574



  // load stt statistics block for all stt-blocks, to decide if the data of queried table exists in current stt file










575



  TStatisBlkArray *pStatisBlkArray = NULL;




170,654,061✔





576



  code = tsdbSttFileReadStatisBlk(pIter->pReader, (const TStatisBlkArray **)&pStatisBlkArray);




170,666,527✔





577



  if (code != TSDB_CODE_SUCCESS) {




170,683,924✔





578



    tsdbError("failed to load stt block statistics, code:%s, %s", tstrerror(code), idStr);




×







579



    return code;




×







580



  }










581












582



  // load statistics block for all tables in current stt file










583



  code = loadSttStatisticsBlockData(pIter->pReader, pIter->pBlockLoadInfo, pStatisBlkArray, suid, idStr);




170,683,924✔





584



  if (code != TSDB_CODE_SUCCESS) {




170,668,388✔





585



    tsdbError("failed to load stt statistics block data, code:%s, %s", tstrerror(code), idStr);




×







586



    return code;




×







587



  }










588












589



  code = loadTombFn(pReader1, pIter->pReader, pIter->pBlockLoadInfo);




170,668,388✔





590












591



  double el = (taosGetTimestampUs() - st) / 1000.0;




170,743,649✔





592



  tsdbDebug("load the stt file blk info completed, elapsed time:%.2fms, %s", el, idStr);




170,743,649✔





593



  return code;




170,712,785✔





594



}










595












596



static int32_t uidComparFn(const void *p1, const void *p2) {




248,329,168✔





597



  const uint64_t *pFirst = p1;




248,329,168✔





598



  const uint64_t *pVal = p2;




248,329,168✔





599












600



  if (*pFirst == *pVal) {




248,329,168✔





601



    return 0;




127,724,969✔





602



  } else {










603



    return *pFirst < *pVal ? -1 : 1;




120,660,565✔





604



  }










605



}










606












607



static void setSttInfoForCurrentTable(SSttBlockLoadInfo *pLoadInfo, uint64_t uid, SSttKeyRange *pRange,




236,543,008✔





608



                                      int64_t *numOfRows) {










609



  if (pRange == NULL || taosArrayGetSize(pLoadInfo->info.pUid) == 0) {




236,543,008✔





610



    return;




48,927,176✔





611



  }










612












613



  int32_t index = taosArraySearchIdx(pLoadInfo->info.pUid, &uid, uidComparFn, TD_EQ);




187,623,185✔





614



  if (index >= 0) {




187,576,952✔





615



    pRange->skey.ts = *(int64_t *)taosArrayGet(pLoadInfo->info.pFirstTs, index);




127,700,708✔





616



    pRange->ekey.ts = *(int64_t *)taosArrayGet(pLoadInfo->info.pLastTs, index);




127,715,727✔





617












618



    *numOfRows += *(int64_t *)taosArrayGet(pLoadInfo->info.pCount, index);




127,702,504✔





619












620



    if (pRange->skey.numOfPKs > 0) {




127,686,814✔





621



      memcpy(&pRange->skey.pks[0], taosArrayGet(pLoadInfo->info.pFirstKey, index), sizeof(SValue));




3,506,304✔





622



      memcpy(&pRange->ekey.pks[0], taosArrayGet(pLoadInfo->info.pLastKey, index), sizeof(SValue));




3,507,255✔





623



    }










624



  }










625



}










626












627



int32_t tLDataIterOpen2(SLDataIter *pIter, SSttFileReader *pSttFileReader, int32_t cid, int8_t backward,




236,572,080✔





628



                        SMergeTreeConf *pConf, SSttBlockLoadInfo *pBlockLoadInfo, SSttKeyRange *pKeyRange,










629



                        int64_t *numOfRows, const char *idStr) {










630



  int32_t code = TSDB_CODE_SUCCESS;




236,572,080✔





631












632



  pIter->uid = pConf->uid;




236,572,080✔





633



  pIter->cid = cid;




236,600,747✔





634



  pIter->backward = backward;




236,624,677✔





635



  pIter->verRange.minVer = pConf->verRange.minVer;




236,600,635✔





636



  pIter->verRange.maxVer = pConf->verRange.maxVer;




236,572,406✔





637



  pIter->timeWindow.skey = pConf->timewindow.skey;




236,595,644✔





638



  pIter->timeWindow.ekey = pConf->timewindow.ekey;




236,558,434✔





639












640



  pIter->pStartRowKey = pConf->pCurRowKey;




236,537,956✔





641



  pIter->pReader = pSttFileReader;




236,510,597✔





642



  pIter->pBlockLoadInfo = pBlockLoadInfo;




236,600,347✔





643












644



  // open stt file failed, ignore and continue










645



  if (pIter->pReader == NULL) {




236,557,495✔





646



    tsdbError("stt file reader is null, %s", idStr);




×







647



    pIter->pSttBlk = NULL;




×







648



    pIter->iSttBlk = -1;




×







649



    return TSDB_CODE_SUCCESS;




×







650



  }










651












652



  if (!pBlockLoadInfo->sttBlockLoaded) {




236,488,739✔





653



    code = doLoadSttFilesBlk(pBlockLoadInfo, pIter, pConf->suid, pConf->loadTombFn, pConf->pReader, idStr);




170,648,795✔





654



    if (code != TSDB_CODE_SUCCESS) {




170,708,474✔





655



      return code;




×







656



    }










657



  }










658












659



  setSttInfoForCurrentTable(pBlockLoadInfo, pConf->uid, pKeyRange, numOfRows);




236,561,242✔





660












661



  // find the start block, actually we could load the position to avoid repeatly searching for the start position when










662



  // the skey is updated.










663



  size_t size = taosArrayGetSize(pBlockLoadInfo->aSttBlk);




236,480,963✔





664



  pIter->iSttBlk = binarySearchForStartBlock(pBlockLoadInfo->aSttBlk->pData, size, pConf->uid, backward);




236,552,093✔





665



  if (pIter->iSttBlk != -1) {




236,511,019✔





666



    pIter->pSttBlk = taosArrayGet(pBlockLoadInfo->aSttBlk, pIter->iSttBlk);




127,782,662✔





667



    pIter->iRow = (pIter->backward) ? pIter->pSttBlk->nRow : -1;




127,762,333✔





668












669



    if ((!backward) && ((pConf->strictTimeRange && pIter->pSttBlk->minKey >= pIter->timeWindow.ekey) ||




127,778,931✔





670



                        (!pConf->strictTimeRange && pIter->pSttBlk->minKey > pIter->timeWindow.ekey))) {




121,281,668✔





671



      pIter->pSttBlk = NULL;




99,104✔





672



    }










673












674



    if (backward && ((pConf->strictTimeRange && pIter->pSttBlk->maxKey <= pIter->timeWindow.skey) ||




127,726,043✔





675



                     (!pConf->strictTimeRange && pIter->pSttBlk->maxKey < pIter->timeWindow.skey))) {




6,497,093✔





676



      pIter->pSttBlk = NULL;




191,573✔





677



      pIter->ignoreEarlierTs = true;




184,340✔





678



    }










679



  }










680












681



  return code;




236,468,020✔





682



}










683












684



void tLDataIterClose2(SLDataIter *pIter) {




170,704,438✔





685



  tsdbSttFileReaderClose(&pIter->pReader);




170,704,438✔





686



  pIter->pReader = NULL;




170,683,102✔





687



}




170,692,103✔





688












689



void tLDataIterNextBlock(SLDataIter *pIter, const char *idStr) {




131,844,278✔





690



  int32_t step = pIter->backward ? -1 : 1;




131,844,278✔





691



  int32_t oldIndex = pIter->iSttBlk;




131,879,337✔





692












693



  pIter->iSttBlk += step;




131,891,796✔





694












695



  int32_t index = -1;




131,875,782✔





696



  size_t  size = pIter->pBlockLoadInfo->aSttBlk->size;




131,875,782✔





697



  for (int32_t i = pIter->iSttBlk; i < size && i >= 0; i += step) {




132,209,176✔





698



    SSttBlk *p = taosArrayGet(pIter->pBlockLoadInfo->aSttBlk, i);




11,974,145✔





699



    if ((!pIter->backward) && p->minUid > pIter->uid) {




11,974,215✔





700



      break;




4,961,617✔





701



    }










702












703



    if (pIter->backward && p->maxUid < pIter->uid) {




7,013,938✔





704



      break;




756,320✔





705



    }










706












707



    // check uid firstly










708



    if (p->minUid <= pIter->uid && p->maxUid >= pIter->uid) {




6,255,903✔





709



      if ((!pIter->backward) && p->minKey > pIter->timeWindow.ekey) {




6,256,007✔





710



        break;




82,886✔





711



      }










712












713



      if (pIter->backward && p->maxKey < pIter->timeWindow.skey) {




6,172,412✔





714



        break;




122✔





715



      }










716












717



      // check time range secondly










718



      if (p->minKey <= pIter->timeWindow.ekey && p->maxKey >= pIter->timeWindow.skey) {




6,172,290✔





719



        if ((!pIter->backward) && p->minVer > pIter->verRange.maxVer) {




5,838,293✔





720



          break;




×







721



        }










722












723



        if (pIter->backward && p->maxVer < pIter->verRange.minVer) {




5,838,397✔





724



          break;




×







725



        }










726












727



        if (p->minVer <= pIter->verRange.maxVer && p->maxVer >= pIter->verRange.minVer) {




5,838,397✔





728



          index = i;




5,837,692✔





729



          break;




5,837,692✔





730



        }










731



      }










732



    }










733



  }










734












735



  pIter->pSttBlk = NULL;




131,895,859✔





736



  if (index != -1) {




131,841,318✔





737



    SSttBlk *p = taosArrayGet(pIter->pBlockLoadInfo->aSttBlk, index);




5,838,397✔





738












739



    pIter->iSttBlk = index;




5,838,293✔





740



    pIter->pSttBlk = (SSttBlk *)taosArrayGet(pIter->pBlockLoadInfo->aSttBlk, pIter->iSttBlk);




5,838,293✔





741



    tsdbDebug("try next stt-file block:%d from %d, trigger by uid:%" PRIu64 ", stt-fileVer:%" PRId64




5,837,588✔





742



              ", uidRange:%" PRId64 "-%" PRId64 " %s",










743



              pIter->iSttBlk, oldIndex, pIter->uid, pIter->cid, p->minUid, p->maxUid, idStr);










744



  } else {










745



    tsdbDebug("no more last block qualified, uid:%" PRIu64 ", stt-file block:%d, %s", pIter->uid, oldIndex, idStr);




126,002,921✔





746



  }










747



}




131,840,509✔





748












749



static int32_t findNextValidRow(SLDataIter *pIter, const char *idStr) {




2,147,483,647✔





750



  bool        hasVal = false;




2,147,483,647✔





751



  int32_t     step = pIter->backward ? -1 : 1;




2,147,483,647✔





752



  int32_t     i = pIter->iRow;




2,147,483,647✔





753



  SBlockData *pData = NULL;




2,147,483,647✔





754












755



  int32_t code = loadLastBlock(pIter, idStr, &pData);




2,147,483,647✔





756



  if (code) {




2,147,483,647✔





757



    tsdbError("failed to load stt block, code:%s, %s", tstrerror(code), idStr);




×







758



    return code;




×







759



  }










760












761



  // mostly we only need to find the start position for a given table










762



  if ((((i == 0) && (!pIter->backward)) || (i == pData->nRow - 1 && pIter->backward)) && pData->aUid != NULL) {




2,147,483,647✔





763



    i = binarySearchForStartRowIndex((uint64_t *)pData->aUid, pData->nRow, pIter->uid, pIter->backward);




128,142,754✔





764



    if (i == -1) {




128,070,434✔





765



      tsdbDebug("failed to find the data in pBlockData, uid:%" PRIu64 " , %s", pIter->uid, idStr);




67,571✔





766



      pIter->iRow = -1;




67,571✔





767



      return code;




67,571✔





768



    }










769



  }










770












771



  for (; i < pData->nRow && i >= 0; i += step) {




2,147,483,647✔





772



    if (pData->aUid != NULL) {




2,147,483,647✔





773



      if (!pIter->backward) {




2,147,483,647✔





774



        if (pData->aUid[i] > pIter->uid) {




2,147,483,647✔





775



          break;




22,593,285✔





776



        }










777



      } else {










778



        if (pData->aUid[i] < pIter->uid) {




500,677,988✔





779



          break;




2,147,783✔





780



        }










781



      }










782



    }










783












784



    int64_t ts = pData->aTSKEY[i];




2,147,483,647✔





785



    if (!pIter->backward) {               // asc




2,147,483,647✔





786



      if (ts > pIter->timeWindow.ekey) {  // no more data




2,147,483,647✔





787



        break;




531,891✔





788



      } else {










789



        if (ts < pIter->timeWindow.skey) {




2,147,483,647✔





790



          continue;




1,105,961,256✔





791



        }










792












793



        if (ts == pIter->timeWindow.skey && pIter->pStartRowKey->numOfPKs > 0) {




2,147,483,647✔





794



          SRowKey key;




158,821✔





795



          tColRowGetKey(pData, i, &key);




158,821✔





796



          int32_t ret = pkCompEx(&key, pIter->pStartRowKey);




158,821✔





797



          if (ret < 0) {




158,821✔





798



            continue;




×







799



          }










800



        }










801



      }










802



    } else {










803



      if (ts < pIter->timeWindow.skey) {




514,580,510✔





804



        break;




792,909✔





805



      } else {










806



        if (ts > pIter->timeWindow.ekey) {




513,777,172✔





807



          continue;




83,385,036✔





808



        }










809












810



        if (ts == pIter->timeWindow.ekey && pIter->pStartRowKey->numOfPKs > 0) {




430,399,513✔





811



          SRowKey key;




×







812



          tColRowGetKey(pData, i, &key);




×







813



          int32_t ret = pkCompEx(&key, pIter->pStartRowKey);




×







814



          if (ret > 0) {




×







815



            continue;




×







816



          }










817



        }










818



      }










819



    }










820












821



    int64_t ver = pData->aVersion[i];




2,147,483,647✔





822



    if (ver < pIter->verRange.minVer) {




2,147,483,647✔





823



      continue;




×







824



    }










825












826



    // todo opt handle desc case










827



    if (ver > pIter->verRange.maxVer) {




2,147,483,647✔





828



      continue;




×







829



    }










830












831



    hasVal = true;




2,147,483,647✔





832



    break;




2,147,483,647✔





833



  }










834












835



  pIter->iRow = (hasVal) ? i : -1;




2,147,483,647✔





836



  return code;




2,147,483,647✔





837



}










838












839



int32_t tLDataIterNextRow(SLDataIter *pIter, const char *idStr, bool *hasNext) {




2,147,483,647✔





840



  int32_t     step = pIter->backward ? -1 : 1;




2,147,483,647✔





841



  int32_t     code = 0;




2,147,483,647✔





842



  int32_t     iBlockL = pIter->iSttBlk;




2,147,483,647✔





843



  SBlockData *pBlockData = NULL;




2,147,483,647✔





844



  int32_t     lino = 0;




2,147,483,647✔





845












846



  *hasNext = false;




2,147,483,647✔





847












848



  // no qualified last file block in current file, no need to fetch row










849



  if (pIter->pSttBlk == NULL) {




2,147,483,647✔





850



    return code;




109,045,174✔





851



  }










852












853



  code = loadLastBlock(pIter, idStr, &pBlockData);




2,147,483,647✔





854



  if (pBlockData == NULL || code != TSDB_CODE_SUCCESS) {




2,147,483,647✔





855



    lino = __LINE__;




1,453✔





856



    goto _exit;




1,453✔





857



  }










858












859



  pIter->iRow += step;




2,147,483,647✔





860












861



  while (1) {




5,838,397✔





862



    bool skipBlock = false;




2,147,483,647✔





863



    code = findNextValidRow(pIter, idStr);




2,147,483,647✔





864



    TSDB_CHECK_CODE(code, lino, _exit);




2,147,483,647✔





865












866



    if (pIter->pBlockLoadInfo->checkRemainingRow) {




2,147,483,647✔





867



      skipBlock = true;




×







868



      int16_t *aCols = pIter->pBlockLoadInfo->colIds;




×







869



      int      nCols = pIter->pBlockLoadInfo->numOfCols;




×







870



      bool     isLast = pIter->pBlockLoadInfo->isLast;




×







871



      for (int inputColIndex = 0; inputColIndex < nCols; ++inputColIndex) {




×







872



        for (int colIndex = 0; colIndex < pBlockData->nColData; ++colIndex) {




×







873



          SColData *pColData = &pBlockData->aColData[colIndex];




×







874



          int16_t   cid = pColData->cid;




×







875












876



          if (cid == aCols[inputColIndex]) {




×







877



            if (isLast && (pColData->flag & HAS_VALUE)) {




×







878



              skipBlock = false;




×







879



              break;




×







880



            } else if (pColData->flag & (HAS_VALUE | HAS_NULL)) {




×







881



              skipBlock = false;




×







882



              break;




×







883



            }










884



          }










885



        }










886



      }










887



    }










888












889



    if (skipBlock || pIter->iRow >= pBlockData->nRow || pIter->iRow < 0) {




2,147,483,647✔





890



      tLDataIterNextBlock(pIter, idStr);




138,090,382✔





891



      if (pIter->pSttBlk == NULL) {  // no more data




131,843,867✔





892



        goto _exit;




126,041,807✔





893



      }










894



    } else {










895



      break;










896



    }










897












898



    if (iBlockL != pIter->iSttBlk) {




5,837,588✔





899



      code = loadLastBlock(pIter, idStr, &pBlockData);




5,837,588✔





900



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




5,838,397✔





901



        lino = __LINE__;




×







902



        goto _exit;




×







903



      }










904












905



      // set start row index










906



      pIter->iRow = pIter->backward ? pBlockData->nRow - 1 : 0;




5,838,397✔





907



    }










908



  }










909












910



  pIter->rInfo.suid = pBlockData->suid;




2,147,483,647✔





911



  pIter->rInfo.uid = pBlockData->uid;




2,147,483,647✔





912



  pIter->rInfo.row = tsdbRowFromBlockData(pBlockData, pIter->iRow);




2,147,483,647✔





913












914



_exit:




2,147,483,647✔





915



  if (code) {




2,147,483,647✔





916



    tsdbError("failed to exec stt-file nextIter, lino:%d, code:%s, %s", lino, tstrerror(code), idStr);




×







917



  }










918












919



  *hasNext = (code == TSDB_CODE_SUCCESS) && (pIter->pSttBlk != NULL) && (pBlockData != NULL);




2,147,483,647✔





920



  return code;




2,147,483,647✔





921



}










922












923



// SMergeTree =================================================










924



static FORCE_INLINE int32_t tLDataIterCmprFn(const SRBTreeNode *p1, const SRBTreeNode *p2) {




2,147,483,647✔





925



  SLDataIter *pIter1 = (SLDataIter *)(((uint8_t *)p1) - offsetof(SLDataIter, node));




2,147,483,647✔





926



  SLDataIter *pIter2 = (SLDataIter *)(((uint8_t *)p2) - offsetof(SLDataIter, node));




2,147,483,647✔





927












928



  SRowKey rkey1 = {0}, rkey2 = {0};




2,147,483,647✔





929



  tRowGetKeyEx(&pIter1->rInfo.row, &rkey1);




2,147,483,647✔





930



  tRowGetKeyEx(&pIter2->rInfo.row, &rkey2);




2,147,483,647✔





931












932



  int32_t ret = tRowKeyCompare(&rkey1, &rkey2);




2,147,483,647✔





933



  if (ret < 0) {




2,147,483,647✔





934



    return -1;




1,364,212,033✔





935



  } else if (ret > 0) {




2,147,483,647✔





936



    return 1;




1,826,580,855✔





937



  } else {










938



    int64_t ver1 = TSDBROW_VERSION(&pIter1->rInfo.row);




2,147,483,647✔





939



    int64_t ver2 = TSDBROW_VERSION(&pIter2->rInfo.row);




2,147,483,647✔





940












941



    if (ver1 < ver2) {




2,147,483,647✔





942



      return -1;




1,214,928,498✔





943



    } else if (ver1 > ver2) {




1,827,680,527✔





944



      return 1;




1,827,686,302✔





945



    } else {










946



      return 0;




×







947



    }










948



  }










949



}










950












951



static FORCE_INLINE int32_t tLDataIterDescCmprFn(const SRBTreeNode *p1, const SRBTreeNode *p2) {




26,568,783✔





952



  return -1 * tLDataIterCmprFn(p1, p2);




26,568,783✔





953



}










954












955



int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoForTable *pSttDataInfo) {




242,481,311✔





956



  int32_t code = TSDB_CODE_SUCCESS;




242,481,311✔





957












958



  pMTree->pIter = NULL;




242,481,311✔





959



  pMTree->backward = pConf->backward;




242,504,039✔





960



  pMTree->idStr = pConf->idstr;




242,480,935✔





961



  int32_t lino = 0;




242,468,821✔





962












963



  if (!pMTree->backward) {  // asc




242,468,821✔





964



    tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn);




232,181,694✔





965



  } else {  // desc










966



    tRBTreeCreate(&pMTree->rbt, tLDataIterDescCmprFn);




10,283,982✔





967



  }










968












969



  pMTree->ignoreEarlierTs = false;




242,458,157✔





970












971



  // no data exists, go to end










972



  int32_t numOfLevels = ((STFileSet *)pConf->pCurrentFileset)->lvlArr->size;




242,440,468✔





973



  if (numOfLevels == 0) {




242,475,189✔





974



    goto _end;




24,540,909✔





975



  }










976












977



  code = adjustSttDataIters(pConf->pSttFileBlockIterArray, pConf->pCurrentFileset);




217,934,280✔





978



  if (code) {




217,857,616✔





979



    goto _end;




×







980



  }










981












982



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




454,156,228✔





983



    SSttLvl *pSttLevel = ((STFileSet *)pConf->pCurrentFileset)->lvlArr->data[j];




236,235,187✔





984



    SArray  *pList = taosArrayGetP(pConf->pSttFileBlockIterArray, j);




236,283,875✔





985












986



    for (int32_t i = 0; i < TARRAY2_SIZE(pSttLevel->fobjArr); ++i) {  // open all last file




472,881,323✔





987



      SLDataIter *pIter = taosArrayGetP(pList, i);




236,604,469✔





988












989



      SSttFileReader    *pSttFileReader = pIter->pReader;




236,584,176✔





990



      SSttBlockLoadInfo *pLoadInfo = pIter->pBlockLoadInfo;




236,585,794✔





991












992



      // open stt file reader if not opened yet










993



      // if failed to open this stt file, ignore the error and try next one










994



      if (pSttFileReader == NULL) {




236,604,910✔





995



        SSttFileReaderConfig conf = {.tsdb = pConf->pTsdb, .szPage = pConf->pTsdb->pVnode->config.tsdbPageSize};




170,696,309✔





996



        conf.file[0] = *pSttLevel->fobjArr->data[i]->f;




170,686,450✔





997












998



        code = tsdbSttFileReaderOpen(pSttLevel->fobjArr->data[i]->fname, &conf, &pSttFileReader);




170,745,960✔





999



        if (code != TSDB_CODE_SUCCESS) {




170,629,329✔





1000



          tsdbError("open stt file reader error. file name %s, code %s, %s", pSttLevel->fobjArr->data[i]->fname,




×







1001



                    tstrerror(code), pMTree->idStr);










1002



        }










1003



      }










1004












1005



      if (pLoadInfo == NULL) {




236,535,076✔





1006



        code = tCreateSttBlockLoadInfo(pConf->pSchema, pConf->pCols, pConf->numOfCols, &pLoadInfo);




170,712,339✔





1007



        if (code != TSDB_CODE_SUCCESS) {




170,690,049✔





1008



          goto _end;




×







1009



        }










1010



      }










1011












1012



      memset(pIter, 0, sizeof(SLDataIter));




236,512,786✔





1013












1014



      SSttKeyRange range = {.skey.numOfPKs = pConf->pCurRowKey->numOfPKs, .ekey.numOfPKs = pConf->pCurRowKey->numOfPKs};




236,512,786✔





1015



      int64_t      numOfRows = 0;




236,585,652✔





1016



      int64_t      cid = pSttLevel->fobjArr->data[i]->f->cid;




236,601,894✔





1017












1018



      code = tLDataIterOpen2(pIter, pSttFileReader, cid, pMTree->backward, pConf, pLoadInfo, &range, &numOfRows,




236,607,753✔





1019



                             pMTree->idStr);










1020



      if (code != TSDB_CODE_SUCCESS) {




236,490,429✔





1021



        goto _end;




×







1022



      }










1023












1024



      bool hasVal = NULL;




236,490,429✔





1025



      code = tLDataIterNextRow(pIter, pMTree->idStr, &hasVal);




236,538,601✔





1026



      if (code) {




236,411,762✔





1027



        goto _end;




×







1028



      }










1029












1030



      if (hasVal) {




236,411,762✔





1031



        tMergeTreeAddIter(pMTree, pIter);




126,551,410✔





1032












1033



        // let's record the time window for current table of uid in the stt files










1034



        if (pSttDataInfo != NULL && numOfRows > 0) {




126,567,151✔





1035



          void *px = taosArrayPush(pSttDataInfo->pKeyRangeList, &range);




126,576,425✔





1036



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




126,591,741✔





1037












1038



          pSttDataInfo->numOfRows += numOfRows;




126,591,741✔





1039



        }










1040



      } else {










1041



        if (!pMTree->ignoreEarlierTs) {




109,860,352✔





1042



          pMTree->ignoreEarlierTs = pIter->ignoreEarlierTs;




109,833,061✔





1043



        }










1044



      }










1045



    }










1046



  }










1047












1048



  return code;




217,921,041✔





1049












1050



_end:




24,543,371✔





1051



  tMergeTreeClose(pMTree);




24,543,371✔





1052



  return code;




24,540,334✔





1053



}










1054












1055



void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) {




126,555,965✔





1056



  SRBTreeNode *node = tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pIter);




126,555,965✔





1057



}




126,572,665✔





1058












1059



bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree) { return pMTree->ignoreEarlierTs; }




×







1060












1061



static void tLDataIterPinSttBlock(SLDataIter *pIter, const char *id) {




2,147,483,647✔





1062



  SSttBlockLoadInfo *pInfo = pIter->pBlockLoadInfo;




2,147,483,647✔





1063












1064



  if (pInfo->blockData[0].sttBlockIndex == pIter->iSttBlk) {




2,147,483,647✔





1065



    pInfo->blockData[0].pin = true;




2,147,483,647✔





1066



    tsdbTrace("pin stt-block, blockIndex:%d, stt-fileVer:%" PRId64 " %s", pIter->iSttBlk, pIter->cid, id);




2,147,483,647✔





1067



    return;




2,147,483,647✔





1068



  }










1069












1070



  if (pInfo->blockData[1].sttBlockIndex == pIter->iSttBlk) {




2,147,483,647✔





1071



    pInfo->blockData[1].pin = true;




2,147,483,647✔





1072



    tsdbTrace("pin stt-block, blockIndex:%d, stt-fileVer:%" PRId64 " %s", pIter->iSttBlk, pIter->cid, id);




2,147,483,647✔





1073



    return;




2,147,483,647✔





1074



  }










1075












1076



  tsdbError("failed to pin any stt block, sttBlock:%d stt-fileVer:%" PRId64 " %s", pIter->iSttBlk, pIter->cid, id);




×







1077



}










1078












1079



static void tLDataIterUnpinSttBlock(SLDataIter *pIter, const char *id) {




2,147,483,647✔





1080



  SSttBlockLoadInfo *pInfo = pIter->pBlockLoadInfo;




2,147,483,647✔





1081



  if (pInfo->blockData[0].pin) {




2,147,483,647✔





1082



    pInfo->blockData[0].pin = false;




2,147,483,647✔





1083



    tsdbTrace("unpin stt-block:%d, stt-fileVer:%" PRId64 " %s", pInfo->blockData[0].sttBlockIndex, pIter->cid, id);




2,147,483,647✔





1084



    return;




2,147,483,647✔





1085



  }










1086












1087



  if (pInfo->blockData[1].pin) {




2,147,483,647✔





1088



    pInfo->blockData[1].pin = false;




2,147,483,647✔





1089



    tsdbTrace("unpin stt-block:%d, stt-fileVer:%" PRId64 " %s", pInfo->blockData[1].sttBlockIndex, pIter->cid, id);




2,147,483,647✔





1090



    return;




2,147,483,647✔





1091



  }










1092












1093



  tsdbError("failed to unpin any stt block, sttBlock:%d stt-fileVer:%" PRId64 " %s", pIter->iSttBlk, pIter->cid, id);




×







1094



}










1095












1096



void tMergeTreePinSttBlock(SMergeTree *pMTree) {




2,147,483,647✔





1097



  if (pMTree->pIter == NULL) {




2,147,483,647✔





1098



    return;




×







1099



  }










1100












1101



  SLDataIter *pIter = pMTree->pIter;




2,147,483,647✔





1102



  pMTree->pPinnedBlockIter = pIter;




2,147,483,647✔





1103



  tLDataIterPinSttBlock(pIter, pMTree->idStr);




2,147,483,647✔





1104



}










1105












1106



void tMergeTreeUnpinSttBlock(SMergeTree *pMTree) {




2,147,483,647✔





1107



  if (pMTree->pPinnedBlockIter == NULL) {




2,147,483,647✔





1108



    return;




×







1109



  }










1110












1111



  SLDataIter *pIter = pMTree->pPinnedBlockIter;




2,147,483,647✔





1112



  pMTree->pPinnedBlockIter = NULL;




2,147,483,647✔





1113



  tLDataIterUnpinSttBlock(pIter, pMTree->idStr);




2,147,483,647✔





1114



}










1115












1116



int32_t tMergeTreeNext(SMergeTree *pMTree, bool *pHasNext) {




2,147,483,647✔





1117



  int32_t code = 0;




2,147,483,647✔





1118



  if (pHasNext == NULL) {




2,147,483,647✔





1119



    return TSDB_CODE_INVALID_PARA;




×







1120



  }










1121












1122



  *pHasNext = false;




2,147,483,647✔





1123



  while (pMTree->pIter) {




2,147,483,647✔





1124



    SLDataIter *pIter = pMTree->pIter;




2,147,483,647✔





1125



    bool        hasVal = false;




2,147,483,647✔





1126



    code = tLDataIterNextRow(pIter, pMTree->idStr, &hasVal);




2,147,483,647✔





1127



    if (!hasVal || (code != 0)) {




2,147,483,647✔





1128



      if (code == TSDB_CODE_FILE_CORRUPTED) {




122,697,370✔





1129



        code = 0;  // suppress the file corrupt error to enable all queries within this cluster can run without failed.




×







1130



      }










1131












1132



      pMTree->pIter = NULL;




122,697,370✔





1133



    }










1134












1135



    // compare with min in RB Tree










1136



    pIter = (SLDataIter *)tRBTreeMin(&pMTree->rbt);




2,147,483,647✔





1137



    if (pMTree->pIter && pIter) {




2,147,483,647✔





1138



      int32_t c = pMTree->rbt.cmprFn(&pMTree->pIter->node, &pIter->node);




1,355,643,155✔





1139



      if (c > 0) {




1,355,625,263✔





1140



        (void)tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);




1,197,149,718✔





1141



        pMTree->pIter = NULL;




1,197,157,638✔





1142



      } else if (!c) {




158,475,545✔





1143



        continue;




×







1144



      }










1145



    }










1146












1147



    break;




2,147,483,647✔





1148



  }










1149












1150



  if (pMTree->pIter == NULL) {




2,147,483,647✔





1151



    pMTree->pIter = (SLDataIter *)tRBTreeMin(&pMTree->rbt);




1,563,262,146✔





1152



    if (pMTree->pIter) {




1,563,228,964✔





1153



      tRBTreeDrop(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);




1,322,301,177✔





1154



    }










1155



  }










1156












1157



  *pHasNext = (pMTree->pIter != NULL);




2,147,483,647✔





1158



  return code;




2,147,483,647✔





1159



}










1160












1161



void tMergeTreeClose(SMergeTree *pMTree) {




527,160,539✔





1162



  pMTree->pIter = NULL;




527,160,539✔





1163



  pMTree->pPinnedBlockIter = NULL;




527,269,628✔





1164



}




527,265,662✔



















    

  • 

    •