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/stream/src/streamSessionState.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 "tstreamFileState.h"










17












18



#include "query.h"










19



#include "streamBackendRocksdb.h"










20



#include "tcommon.h"










21



#include "tsimplehash.h"










22












23



int sessionStateKeyCompare(const void* pWin1, const void* pDatas, int pos) {




×







24



  SRowBuffPos* pPos2 = taosArrayGetP(pDatas, pos);




×







25



  SSessionKey* pWin2 = (SSessionKey*)pPos2->pKey;




×







26



  return sessionWinKeyCmpr((SSessionKey*)pWin1, pWin2);




×







27



}










28












29



int sessionStateRangeKeyCompare(const SSessionKey* pWin1, const void* pDatas, int pos) {




×







30



  SRowBuffPos* pPos2 = taosArrayGetP(pDatas, pos);




×







31



  SSessionKey* pWin2 = (SSessionKey*)pPos2->pKey;




×







32



  return sessionRangeKeyCmpr(pWin1, pWin2);




×







33



}










34












35



int32_t binarySearch(void* keyList, int num, const void* key, __session_compare_fn_t cmpFn) {




×







36



  int firstPos = 0, lastPos = num - 1, midPos = -1;




×







37



  int numOfRows = 0;




×







38












39



  if (num <= 0) return -1;





×







40



  // find the first position which is smaller or equal than the key.










41



  // if all data is bigger than the key return -1










42



  while (1) {










43



    if (cmpFn(key, keyList, lastPos) >= 0) return lastPos;





×







44



    if (cmpFn(key, keyList, firstPos) == 0) return firstPos;





×







45



    if (cmpFn(key, keyList, firstPos) < 0) return firstPos - 1;





×







46












47



    numOfRows = lastPos - firstPos + 1;




×







48



    midPos = (numOfRows >> 1) + firstPos;




×







49












50



    if (cmpFn(key, keyList, midPos) < 0) {





×







51



      lastPos = midPos - 1;




×







52



    } else if (cmpFn(key, keyList, midPos) > 0) {





×







53



      firstPos = midPos + 1;




×







54



    } else {










55



      break;




×







56



    }










57



  }










58












59



  return midPos;




×







60



}










61












62



int64_t getSessionWindowEndkey(void* data, int32_t index) {




×







63



  SArray*       pWinInfos = (SArray*)data;




×







64



  SRowBuffPos** ppos = taosArrayGet(pWinInfos, index);




×







65



  if (ppos != NULL) {





×







66



    SSessionKey* pWin = (SSessionKey*)((*ppos)->pKey);




×







67



    return pWin->win.ekey;




×







68



  } else {










69



    return 0;




×







70



  }










71



}










72












73



bool inSessionWindow(SSessionKey* pKey, TSKEY ts, int64_t gap) {




×







74



  if (ts + gap >= pKey->win.skey && ts - gap <= pKey->win.ekey) {





×







75



    return true;




×







76



  }










77



  return false;




×







78



}










79












80



SStreamStateCur* createStateCursor(SStreamFileState* pFileState) {




×







81



  SStreamStateCur* pCur = createStreamStateCursor();




×







82



  if (pCur == NULL) {





×







83



    return NULL;




×







84



  }










85



  pCur->pStreamFileState = pFileState;




×







86



  return pCur;




×







87



}










88












89



static int32_t addNewSessionWindow(SStreamFileState* pFileState, SArray* pWinInfos, const SSessionKey* pKey,




×







90



                                   SRowBuffPos** ppPos) {










91



  int32_t      code = TSDB_CODE_SUCCESS;




×







92



  int32_t      lino = 0;




×







93



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×







94



  if (!pNewPos || !pNewPos->pRowBuff) {





×







95



    code = TSDB_CODE_OUT_OF_MEMORY;




×







96



    QUERY_CHECK_CODE(code, lino, _end);





×







97



  }










98












99



  memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));




×







100



  void* tmp = taosArrayPush(pWinInfos, &pNewPos);




×







101



  if (!tmp) {





×







102



    code = terrno;




×







103



    QUERY_CHECK_CODE(code, lino, _end);





×







104



  }










105



  (*ppPos) = pNewPos;




×







106












107



_end:




×







108



  if (code != TSDB_CODE_SUCCESS) {





×







109



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





×







110



  }










111



  return code;




×







112



}










113












114



static int32_t insertNewSessionWindow(SStreamFileState* pFileState, SArray* pWinInfos, const SSessionKey* pKey,




×







115



                                      int32_t index, SRowBuffPos** ppPos) {










116



  int32_t      code = TSDB_CODE_SUCCESS;




×







117



  int32_t      lino = 0;




×







118



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×







119



  if (!pNewPos || !pNewPos->pRowBuff) {





×







120



    code = TSDB_CODE_OUT_OF_MEMORY;




×







121



    QUERY_CHECK_CODE(code, lino, _end);





×







122



  }










123












124



  memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));




×







125



  void* tmp = taosArrayInsert(pWinInfos, index, &pNewPos);




×







126



  if (!tmp) {





×







127



    code = terrno;




×







128



    QUERY_CHECK_CODE(code, lino, _end);





×







129



  }










130












131



  *ppPos = pNewPos;




×







132












133



_end:




×







134



  if (code != TSDB_CODE_SUCCESS) {





×







135



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





×







136



  }










137



  return code;




×







138



}










139












140



SRowBuffPos* createSessionWinBuff(SStreamFileState* pFileState, SSessionKey* pKey, void* p, int32_t* pVLen) {




×







141



  int32_t      code = TSDB_CODE_SUCCESS;




×







142



  int32_t      lino = 0;




×







143



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×







144



  if (!pNewPos || !pNewPos->pRowBuff) {





×







145



    code = TSDB_CODE_OUT_OF_MEMORY;




×







146



    QUERY_CHECK_CODE(code, lino, _end);





×







147



  }










148












149



  memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));




×







150



  pNewPos->needFree = true;




×







151



  pNewPos->beFlushed = true;




×







152



  if (p) {





×







153



    memcpy(pNewPos->pRowBuff, p, *pVLen);




×







154



  } else {










155



    int32_t len = getRowStateRowSize(pFileState);




×







156



    memset(pNewPos->pRowBuff, 0, len);




×







157



  }










158












159



_end:




×







160



  taosMemoryFree(p);





×







161



  if (code != TSDB_CODE_SUCCESS) {





×







162



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





×







163



    return NULL;




×







164



  }










165



  return pNewPos;




×







166



}










167












168



int32_t getSessionWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, TSKEY gap, void** pVal, int32_t* pVLen,




×







169



                                int32_t* pWinCode) {










170



  int32_t code = TSDB_CODE_SUCCESS;




×







171



  int32_t lino = 0;




×







172



  (*pWinCode) = TSDB_CODE_SUCCESS;




×







173



  SSHashObj* pSessionBuff = getRowStateBuff(pFileState);




×







174



  SArray*    pWinStates = NULL;




×







175



  void**     ppBuff = tSimpleHashGet(pSessionBuff, &pKey->groupId, sizeof(uint64_t));




×







176



  if (ppBuff) {





×







177



    pWinStates = (SArray*)(*ppBuff);




×







178



  } else {










179



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







180



    if (!pWinStates) {





×







181



      code = terrno;




×







182



      QUERY_CHECK_CODE(code, lino, _end);





×







183



    }










184



    code = tSimpleHashPut(pSessionBuff, &pKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);




×







185



    QUERY_CHECK_CODE(code, lino, _end);





×







186



  }










187












188



  TSKEY startTs = pKey->win.skey;




×







189



  TSKEY endTs = pKey->win.ekey;




×







190












191



  int32_t size = taosArrayGetSize(pWinStates);




×







192



  if (size == 0) {





×







193



    void*   pFileStore = getStateFileStore(pFileState);




×







194



    void*   p = NULL;




×







195



    int32_t code_file = streamStateSessionAddIfNotExist_rocksdb(pFileStore, pKey, gap, &p, pVLen);




×







196



    if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {





×







197



      (*pVal) = createSessionWinBuff(pFileState, pKey, p, pVLen);




×







198



      if (!(*pVal)) {





×







199



        code = TSDB_CODE_OUT_OF_MEMORY;




×







200



        QUERY_CHECK_CODE(code, lino, _end);





×







201



      }










202












203



      (*pWinCode) = code_file;




×







204



      qDebug("===stream===0 get session win:%" PRId64 ",%" PRId64 " from disc, res %d", startTs, endTs, code_file);





×







205



    } else {










206



      code = addNewSessionWindow(pFileState, pWinStates, pKey, (SRowBuffPos**)pVal);




×







207



      (*pWinCode) = TSDB_CODE_FAILED;




×







208



      taosMemoryFree(p);





×







209



      QUERY_CHECK_CODE(code, lino, _end);





×







210



    }










211



    goto _end;




×







212



  }










213












214



  // find the first position which is smaller than the pKey










215



  int32_t      index = binarySearch(pWinStates, size, pKey, sessionStateKeyCompare);




×







216



  SRowBuffPos* pPos = NULL;




×







217












218



  if (index >= 0) {





×







219



    pPos = taosArrayGetP(pWinStates, index);




×







220



    if (inSessionWindow(pPos->pKey, startTs, gap)) {





×







221



      (*pVal) = pPos;




×







222



      SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;




×







223



      pPos->beUsed = true;




×







224



      pPos->beFlushed = false;




×







225



      *pKey = *pDestWinKey;




×







226



      goto _end;




×







227



    }










228



  }










229












230



  if (index + 1 < size) {





×







231



    pPos = taosArrayGetP(pWinStates, index + 1);




×







232



    if (inSessionWindow(pPos->pKey, startTs, gap) || (endTs != INT64_MIN && inSessionWindow(pPos->pKey, endTs, gap))) {





×







233



      (*pVal) = pPos;




×







234



      SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;




×







235



      pPos->beUsed = true;




×







236



      pPos->beFlushed = false;




×







237



      *pKey = *pDestWinKey;




×







238



      goto _end;




×







239



    }










240



  }










241












242



  if (index + 1 == 0) {





×







243



    if (!isDeteled(pFileState, endTs) && isFlushedState(pFileState, endTs, gap)) {





×







244



      void*   p = NULL;




×







245



      void*   pFileStore = getStateFileStore(pFileState);




×







246



      int32_t code_file = streamStateSessionAddIfNotExist_rocksdb(pFileStore, pKey, gap, &p, pVLen);




×







247



      if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {





×







248



        (*pVal) = createSessionWinBuff(pFileState, pKey, p, pVLen);




×







249



        if (!(*pVal)) {





×







250



          code = TSDB_CODE_OUT_OF_MEMORY;




×







251



          QUERY_CHECK_CODE(code, lino, _end);





×







252



        }










253












254



        (*pWinCode) = code_file;




×







255



        qDebug("===stream===1 get session win:%" PRId64 ",%" PRId64 " from disc, res %d", startTs, endTs, code_file);





×







256



        goto _end;




×







257



      } else {










258



        taosMemoryFree(p);





×







259



      }










260



    }










261



  }










262












263



  if (index == size - 1) {





×







264



    code = addNewSessionWindow(pFileState, pWinStates, pKey, (SRowBuffPos**)pVal);




×







265



    QUERY_CHECK_CODE(code, lino, _end);





×







266












267



    (*pWinCode) = TSDB_CODE_FAILED;




×







268



    goto _end;




×







269



  }










270












271



  code = insertNewSessionWindow(pFileState, pWinStates, pKey, index + 1, (SRowBuffPos**)pVal);




×







272



  QUERY_CHECK_CODE(code, lino, _end);





×







273












274



  (*pWinCode) = TSDB_CODE_FAILED;




×







275












276



_end:




×







277



  if (code != TSDB_CODE_SUCCESS) {





×







278



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





×







279



  }










280



  return code;




×







281



}










282












283



int32_t getSessionRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen,




×







284



                          int32_t* pWinCode) {










285



  SWinKey*    pTmpkey = pKey;




×







286



  SSessionKey pWinKey = {.groupId = pTmpkey->groupId, .win.skey = pTmpkey->ts, .win.ekey = pTmpkey->ts};




×







287



  return getSessionWinResultBuff(pFileState, &pWinKey, 0, pVal, pVLen, pWinCode);




×







288



}










289












290



int32_t putSessionWinResultBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {




×







291



  int32_t      code = TSDB_CODE_SUCCESS;




×







292



  int32_t      lino = 0;




×







293



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







294



  SSessionKey* pKey = pPos->pKey;




×







295



  SArray*      pWinStates = NULL;




×







296



  void**       ppBuff = tSimpleHashGet(pSessionBuff, &pKey->groupId, sizeof(uint64_t));




×







297



  if (ppBuff) {





×







298



    pWinStates = (SArray*)(*ppBuff);




×







299



  } else {










300



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







301



    if (!pWinStates) {





×







302



      code = terrno;




×







303



      QUERY_CHECK_CODE(code, lino, _end);





×







304



    }










305












306



    code = tSimpleHashPut(pSessionBuff, &pKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);




×







307



    QUERY_CHECK_CODE(code, lino, _end);





×







308



  }










309












310



  int32_t size = taosArrayGetSize(pWinStates);




×







311



  if (size == 0) {





×







312



    void* tmp = taosArrayPush(pWinStates, &pPos);




×







313



    if (!tmp) {





×







314



      code = terrno;




×







315



      QUERY_CHECK_CODE(code, lino, _end);





×







316



    }










317



    goto _end;




×







318



  }










319












320



  // find the first position which is smaller than the pKey










321



  int32_t index = binarySearch(pWinStates, size, pKey, sessionStateKeyCompare);




×







322



  if (index >= 0) {





×







323



    void* tmp = taosArrayInsert(pWinStates, index, &pPos);




×







324



    if (!tmp) {





×







325



      code = terrno;




×







326



      QUERY_CHECK_CODE(code, lino, _end);





×







327



    }










328



  } else {










329



    void* tmp = taosArrayInsert(pWinStates, 0, &pPos);




×







330



    if (!tmp) {





×







331



      code = terrno;




×







332



      QUERY_CHECK_CODE(code, lino, _end);





×







333



    }










334



  }










335












336



_end:




×







337



  pPos->needFree = false;




×







338



  if (code != TSDB_CODE_SUCCESS) {





×







339



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





×







340



  }










341



  return code;




×







342



}










343












344



int32_t getSessionFlushedBuff(SStreamFileState* pFileState, SSessionKey* pKey, void** pVal, int32_t* pVLen,




×







345



                              int32_t* pWinCode) {










346



  int32_t      code = TSDB_CODE_SUCCESS;




×







347



  int32_t      lino = 0;




×







348



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×







349



  if (!pNewPos || !pNewPos->pRowBuff) {





×







350



    code = TSDB_CODE_OUT_OF_MEMORY;




×







351



    QUERY_CHECK_CODE(code, lino, _end);





×







352



  }










353



  pNewPos->needFree = true;




×







354



  pNewPos->beFlushed = true;




×







355



  void* pBuff = NULL;




×







356



  (*pWinCode) = streamStateSessionGet_rocksdb(getStateFileStore(pFileState), pKey, &pBuff, pVLen);




×







357



  if ((*pWinCode) != TSDB_CODE_SUCCESS) {





×







358



    goto _end;




×







359



  }










360



  memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));




×







361



  memcpy(pNewPos->pRowBuff, pBuff, *pVLen);




×







362



  taosMemoryFreeClear(pBuff);





×







363



  (*pVal) = pNewPos;




×







364












365



_end:




×







366



  if (code != TSDB_CODE_SUCCESS) {





×







367



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





×







368



  }










369



  return code;




×







370



}










371












372



int32_t deleteSessionWinStateBuffFn(void* pBuff, const void* key, size_t keyLen) {




×







373



  SSHashObj*   pSessionBuff = (SSHashObj*)pBuff;




×







374



  SSessionKey* pWinKey = (SSessionKey*)key;




×







375



  void**       ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







376



  if (!ppBuff) {





×







377



    return TSDB_CODE_SUCCESS;




×







378



  }










379



  SArray* pWinStates = (SArray*)(*ppBuff);




×







380



  int32_t size = taosArrayGetSize(pWinStates);




×







381



  TSKEY   gap = 0;




×







382



  int32_t index = binarySearch(pWinStates, size, pWinKey, sessionStateKeyCompare);




×







383



  if (index >= 0) {





×







384



    SRowBuffPos* pPos = taosArrayGetP(pWinStates, index);




×







385



    if (inSessionWindow(pPos->pKey, pWinKey->win.skey, gap)) {





×







386



      pPos->beFlushed = true;




×







387



      taosArrayRemove(pWinStates, index);




×







388



    }










389



  }










390



  return TSDB_CODE_SUCCESS;




×







391



}










392












393



void deleteSessionWinStateBuffByPosFn(SStreamFileState* pFileState, SRowBuffPos* pPos) {




×







394



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







395



  SSessionKey* pWinKey = (SSessionKey*)pPos->pKey;




×







396



  void**       ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







397



  if (!ppBuff) {





×







398



    return;




×







399



  }










400



  SArray* pWinStates = (SArray*)(*ppBuff);




×







401



  int32_t size = taosArrayGetSize(pWinStates);




×







402



  TSKEY   gap = 0;




×







403



  int32_t index = binarySearch(pWinStates, size, pWinKey, sessionStateKeyCompare);




×







404



  if (index >= 0) {





×







405



    SRowBuffPos* pItemPos = taosArrayGetP(pWinStates, index);




×







406



    if (pItemPos == pPos) {





×







407



      taosArrayRemove(pWinStates, index);




×







408



    }










409



  }










410



}










411












412



int32_t allocSessioncWinBuffByNextPosition(SStreamFileState* pFileState, SStreamStateCur* pCur,




×







413



                                           const SSessionKey* pWinKey, void** ppVal, int32_t* pVLen) {










414



  int32_t      code = TSDB_CODE_SUCCESS;




×







415



  int32_t      lino = 0;




×







416



  SRowBuffPos* pNewPos = NULL;




×







417



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







418



  void**       ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







419



  SArray*      pWinStates = NULL;




×







420



  if (!ppBuff) {





×







421



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







422



    if (!pWinStates) {





×







423



      code = terrno;




×







424



      QUERY_CHECK_CODE(code, lino, _end);





×







425



    }










426












427



    code = tSimpleHashPut(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);




×







428



    QUERY_CHECK_CODE(code, lino, _end);





×







429



  } else {










430



    pWinStates = (SArray*)(*ppBuff);




×







431



  }










432



  if (!pCur) {





×







433



    code = addNewSessionWindow(pFileState, pWinStates, pWinKey, &pNewPos);




×







434



    QUERY_CHECK_CODE(code, lino, _end);





×







435












436



    goto _end;




×







437



  }










438












439



  int32_t size = taosArrayGetSize(pWinStates);




×







440



  if (pCur->buffIndex >= 0) {





×







441



    if (pCur->buffIndex >= size) {





×







442



      code = addNewSessionWindow(pFileState, pWinStates, pWinKey, &pNewPos);




×







443



      QUERY_CHECK_CODE(code, lino, _end);





×







444












445



      goto _end;




×







446



    }










447



    code = insertNewSessionWindow(pFileState, pWinStates, pWinKey, pCur->buffIndex, &pNewPos);




×







448



    QUERY_CHECK_CODE(code, lino, _end);





×







449












450



    goto _end;




×







451



  } else {










452



    if (size > 0) {





×







453



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, 0);




×







454



      if (sessionWinKeyCmpr(pWinKey, pPos->pKey) >= 0) {





×







455



        // pCur is invalid










456



        SSessionKey pTmpKey = *pWinKey;




×







457



        int32_t     winCode = TSDB_CODE_SUCCESS;




×







458



        code = getSessionWinResultBuff(pFileState, &pTmpKey, 0, (void**)&pNewPos, pVLen, &winCode);




×







459



        QUERY_CHECK_CONDITION((winCode == TSDB_CODE_FAILED), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);





×







460



        QUERY_CHECK_CODE(code, lino, _end);





×







461



        goto _end;




×







462



      }










463



    }










464



    pNewPos = getNewRowPosForWrite(pFileState);




×







465



    if (!pNewPos || !pNewPos->pRowBuff) {





×







466



      code = TSDB_CODE_OUT_OF_MEMORY;




×







467



      QUERY_CHECK_CODE(code, lino, _end);





×







468



    }










469












470



    memcpy(pNewPos->pKey, pWinKey, sizeof(SSessionKey));




×







471



    pNewPos->needFree = true;




×







472



    pNewPos->beFlushed = true;




×







473



  }










474












475



_end:




×







476



  (*ppVal) = pNewPos;




×







477



  if (code != TSDB_CODE_SUCCESS) {





×







478



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





×







479



  }










480



  return code;




×







481



}










482












483



void sessionWinStateClear(SStreamFileState* pFileState) {




×







484



  int32_t buffSize = getRowStateRowSize(pFileState);




×







485



  void*   pIte = NULL;




×







486



  size_t  keyLen = 0;




×







487



  int32_t iter = 0;




×







488



  void*   pBuff = getRowStateBuff(pFileState);




×







489



  while ((pIte = tSimpleHashIterate(pBuff, pIte, &iter)) != NULL) {





×







490



    SArray* pWinStates = *((void**)pIte);




×







491



    int32_t size = taosArrayGetSize(pWinStates);




×







492



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





×







493



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, i);




×







494



      memset(pPos->pRowBuff, 0, buffSize);




×







495



    }










496



  }










497



}




×







498












499



void sessionWinStateCleanup(void* pBuff) {




×







500



  void*   pIte = NULL;




×







501



  size_t  keyLen = 0;




×







502



  int32_t iter = 0;




×







503



  while ((pIte = tSimpleHashIterate(pBuff, pIte, &iter)) != NULL) {





×







504



    SArray* pWinStates = (SArray*)(*(void**)pIte);




×







505



    taosArrayDestroy(pWinStates);




×







506



  }










507



  tSimpleHashCleanup(pBuff);




×







508



}




×







509












510



static SStreamStateCur* seekKeyCurrentPrev_buff(SStreamFileState* pFileState, const SSessionKey* pWinKey,




×







511



                                                SArray** pWins, int32_t* pIndex) {










512



  SStreamStateCur* pCur = NULL;




×







513



  SSHashObj*       pSessionBuff = getRowStateBuff(pFileState);




×







514



  void**           ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







515



  if (!ppBuff) {





×







516



    return NULL;




×







517



  }










518












519



  SArray* pWinStates = (SArray*)(*ppBuff);




×







520



  int32_t size = taosArrayGetSize(pWinStates);




×







521



  TSKEY   gap = 0;




×







522



  int32_t index = binarySearch(pWinStates, size, pWinKey, sessionStateKeyCompare);




×







523












524



  if (pWins) {





×







525



    (*pWins) = pWinStates;




×







526



  }










527












528



  if (size > 0 && index == -1) {





×







529



    SRowBuffPos* pPos = taosArrayGetP(pWinStates, 0);




×







530



    SSessionKey* pWin = (SSessionKey*)pPos->pKey;




×







531



    if (pWinKey->win.skey == pWin->win.skey) {





×







532



      index = 0;




×







533



    }










534



  }










535












536



  if (index >= 0) {





×







537



    pCur = createStateCursor(pFileState);




×







538



    if (pCur == NULL) {





×







539



      return NULL;




×







540



    }










541



    pCur->buffIndex = index;




×







542



    if (pIndex) {





×







543



      *pIndex = index;




×







544



    }










545



  }










546












547



  return pCur;




×







548



}










549












550



SStreamStateCur* sessionWinStateSeekKeyCurrentPrev(SStreamFileState* pFileState, const SSessionKey* pWinKey) {




×







551



  SStreamStateCur* pCur = seekKeyCurrentPrev_buff(pFileState, pWinKey, NULL, NULL);




×







552



  if (pCur) {





×







553



    return pCur;




×







554



  }










555












556



  void* pFileStore = getStateFileStore(pFileState);




×







557



  pCur = streamStateSessionSeekKeyCurrentPrev_rocksdb(pFileStore, pWinKey);




×







558



  if (!pCur) {





×







559



    return NULL;




×







560



  }










561



  pCur->buffIndex = -1;




×







562



  pCur->pStreamFileState = pFileState;




×







563



  return pCur;




×







564



}










565



static void transformCursor(SStreamFileState* pFileState, SStreamStateCur* pCur) {




×







566



  if (!pCur) {





×







567



    return;




×







568



  }










569



  streamStateResetCur(pCur);




×







570



  pCur->buffIndex = 0;




×







571



  pCur->pStreamFileState = pFileState;




×







572



}










573












574



static void checkAndTransformCursor(SStreamFileState* pFileState, const uint64_t groupId, SArray* pWinStates,




×







575



                                    SStreamStateCur** ppCur) {










576



  SSessionKey key = {.groupId = groupId};




×







577



  int32_t     code = streamStateSessionGetKVByCur_rocksdb(getStateFileStore(pFileState), *ppCur, &key, NULL, NULL);




×







578



  if (taosArrayGetSize(pWinStates) > 0 &&





×







579



      (code == TSDB_CODE_FAILED || sessionStateKeyCompare(&key, pWinStates, 0) >= 0)) {





×







580



    if (!(*ppCur)) {





×







581



      (*ppCur) = createStateCursor(pFileState);




×







582



    }










583



    transformCursor(pFileState, *ppCur);




×







584



  } else if (*ppCur) {





×







585



    (*ppCur)->buffIndex = -1;




×







586



    (*ppCur)->pStreamFileState = pFileState;




×







587



  }










588



}




×







589












590



SStreamStateCur* sessionWinStateSeekKeyCurrentNext(SStreamFileState* pFileState, const SSessionKey* pWinKey) {




×







591



  SArray*          pWinStates = NULL;




×







592



  int32_t          index = -1;




×







593



  SStreamStateCur* pCur = seekKeyCurrentPrev_buff(pFileState, pWinKey, &pWinStates, &index);




×







594



  if (pCur) {





×







595



    if (sessionStateRangeKeyCompare(pWinKey, pWinStates, index) > 0) {





×







596



      sessionWinStateMoveToNext(pCur);




×







597



    }










598



    return pCur;




×







599



  }










600












601



  void* pFileStore = getStateFileStore(pFileState);




×







602



  pCur = streamStateSessionSeekKeyCurrentNext_rocksdb(pFileStore, (SSessionKey*)pWinKey);




×







603



  checkAndTransformCursor(pFileState, pWinKey->groupId, pWinStates, &pCur);




×







604



  return pCur;




×







605



}










606












607



SStreamStateCur* sessionWinStateSeekKeyNext(SStreamFileState* pFileState, const SSessionKey* pWinKey) {




×







608



  SArray*          pWinStates = NULL;




×







609



  int32_t          index = -1;




×







610



  SStreamStateCur* pCur = seekKeyCurrentPrev_buff(pFileState, pWinKey, &pWinStates, &index);




×







611



  if (pCur) {





×







612



    sessionWinStateMoveToNext(pCur);




×







613



    return pCur;




×







614



  }










615












616



  void* pFileStore = getStateFileStore(pFileState);




×







617



  pCur = streamStateSessionSeekKeyNext_rocksdb(pFileStore, pWinKey);




×







618



  checkAndTransformCursor(pFileState, pWinKey->groupId, pWinStates, &pCur);




×







619



  return pCur;




×







620



}










621












622



SStreamStateCur* countWinStateSeekKeyPrev(SStreamFileState* pFileState, const SSessionKey* pWinKey, COUNT_TYPE count) {




×







623



  SArray*          pWinStates = NULL;




×







624



  int32_t          index = -1;




×







625



  SStreamStateCur* pBuffCur = seekKeyCurrentPrev_buff(pFileState, pWinKey, &pWinStates, &index);




×







626



  int32_t          resSize = getRowStateRowSize(pFileState);




×







627



  COUNT_TYPE       winCount = 0;




×







628



  if (pBuffCur) {





×







629



    while (index >= 0) {





×







630



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, index);




×







631



      winCount = *((COUNT_TYPE*)((char*)pPos->pRowBuff + (resSize - sizeof(COUNT_TYPE))));




×







632



      if (sessionStateRangeKeyCompare(pWinKey, pWinStates, index) == 0 || winCount < count) {





×







633



        index--;




×







634



      } else if (index >= 0) {





×







635



        pBuffCur->buffIndex = index + 1;




×







636



        return pBuffCur;




×







637



      }










638



    }










639



    pBuffCur->buffIndex = 0;




×







640



  } else if (taosArrayGetSize(pWinStates) > 0) {





×







641



    pBuffCur = createStateCursor(pFileState);




×







642



    if (pBuffCur == NULL) {





×







643



      return NULL;




×







644



    }










645



    pBuffCur->buffIndex = 0;




×







646



  }










647












648



  void*            pFileStore = getStateFileStore(pFileState);




×







649



  SStreamStateCur* pCur = streamStateSessionSeekKeyPrev_rocksdb(pFileStore, pWinKey);




×







650



  if (pCur) {





×







651



    pCur->pStreamFileState = pFileState;




×







652



    SSessionKey key = {0};




×







653



    void*       pVal = NULL;




×







654



    int         len = 0;




×







655



    int32_t     code = streamStateSessionGetKVByCur_rocksdb(getStateFileStore(pFileState), pCur, &key, &pVal, &len);




×







656



    if (code == TSDB_CODE_FAILED) {





×







657



      streamStateFreeCur(pCur);




×







658



      return pBuffCur;




×







659



    }










660



    winCount = *((COUNT_TYPE*)((char*)pVal + (resSize - sizeof(COUNT_TYPE))));




×







661



    taosMemoryFreeClear(pVal);





×







662



    streamStateFreeCur(pBuffCur);




×







663



    if (sessionRangeKeyCmpr(pWinKey, &key) != 0 && winCount == count) {





×







664



      streamStateCurNext(pFileStore, pCur);




×







665



      return pCur;




×







666



    }










667



    streamStateCurPrev(pFileStore, pCur);




×







668



    while (1) {










669



      code = streamStateSessionGetKVByCur_rocksdb(NULL, pCur, &key, &pVal, &len);




×







670



      if (code == TSDB_CODE_FAILED) {





×







671



        streamStateCurNext(pFileStore, pCur);




×







672



        return pCur;




×







673



      }










674



      winCount = *((COUNT_TYPE*)((char*)pVal + (resSize - sizeof(COUNT_TYPE))));




×







675



      taosMemoryFreeClear(pVal);





×







676



      if (sessionRangeKeyCmpr(pWinKey, &key) == 0 || winCount < count) {





×







677



        streamStateCurPrev(pFileStore, pCur);




×







678



      } else {










679



        streamStateCurNext(pFileStore, pCur);




×







680



        return pCur;




×







681



      }










682



    }










683



  }










684



  return pBuffCur;




×







685



}










686












687



int32_t sessionWinStateGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen) {




×







688



  if (!pCur) {





×







689



    return TSDB_CODE_FAILED;




×







690



  }










691



  int32_t code = TSDB_CODE_SUCCESS;




×







692












693



  SSHashObj* pSessionBuff = getRowStateBuff(pCur->pStreamFileState);




×







694



  void**     ppBuff = tSimpleHashGet(pSessionBuff, &pKey->groupId, sizeof(uint64_t));




×







695



  SArray*    pWinStates = NULL;




×







696



  if (ppBuff) {





×







697



    pWinStates = (SArray*)(*ppBuff);




×







698



  }










699












700



  if (pCur->buffIndex >= 0) {





×







701



    int32_t size = taosArrayGetSize(pWinStates);




×







702



    if (pCur->buffIndex >= size) {





×







703



      return TSDB_CODE_FAILED;




×







704



    }










705



    SRowBuffPos* pPos = taosArrayGetP(pWinStates, pCur->buffIndex);




×







706



    if (pVal) {





×







707



      *pVal = pPos;




×







708



    }










709



    *pKey = *(SSessionKey*)(pPos->pKey);




×







710



  } else {










711



    void* pData = NULL;




×







712



    code = streamStateSessionGetKVByCur_rocksdb(getStateFileStore(pCur->pStreamFileState), pCur, pKey, &pData, pVLen);




×







713



    if (taosArrayGetSize(pWinStates) > 0 &&





×







714



        (code == TSDB_CODE_FAILED || sessionStateRangeKeyCompare(pKey, pWinStates, 0) >= 0)) {





×







715



      transformCursor(pCur->pStreamFileState, pCur);




×







716



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, pCur->buffIndex);




×







717



      if (pVal) {





×







718



        *pVal = pPos;




×







719



      }










720



      *pKey = *(SSessionKey*)(pPos->pKey);




×







721



      code = TSDB_CODE_SUCCESS;




×







722



    } else if (code == TSDB_CODE_SUCCESS && pVal) {





×







723



      SRowBuffPos* pNewPos = getNewRowPosForWrite(pCur->pStreamFileState);




×







724



      if (!pNewPos || !pNewPos->pRowBuff) {





×







725



        code = TSDB_CODE_OUT_OF_MEMORY;




×







726



        taosMemoryFreeClear(pData);





×







727



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





×







728



        return code;




×







729



      }










730



      memcpy(pNewPos->pKey, pKey, sizeof(SSessionKey));




×







731



      pNewPos->needFree = true;




×







732



      pNewPos->beFlushed = true;




×







733



      memcpy(pNewPos->pRowBuff, pData, *pVLen);




×







734



      (*pVal) = pNewPos;




×







735



    }










736



    taosMemoryFreeClear(pData);





×







737



  }










738



  return code;




×







739



}










740












741



void sessionWinStateMoveToNext(SStreamStateCur* pCur) {




×







742



  qTrace("move cursor to next");





×







743



  if (pCur && pCur->buffIndex >= 0) {





×







744



    pCur->buffIndex++;




×







745



  } else {










746



    streamStateCurNext_rocksdb(pCur);




×







747



  }










748



}




×







749












750



int32_t sessionWinStateGetKeyByRange(SStreamFileState* pFileState, const SSessionKey* key, SSessionKey* curKey,




×







751



                                     range_cmpr_fn cmpFn) {










752



  SStreamStateCur* pCur = sessionWinStateSeekKeyCurrentPrev(pFileState, key);




×







753



  SSessionKey      tmpKey = *key;




×







754



  int32_t          code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL);




×







755



  bool             hasCurrentPrev = true;




×







756



  if (code == TSDB_CODE_FAILED) {





×







757



    streamStateFreeCur(pCur);




×







758



    pCur = sessionWinStateSeekKeyNext(pFileState, key);




×







759



    code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL);




×







760



    hasCurrentPrev = false;




×







761



  }










762












763



  if (code == TSDB_CODE_FAILED) {





×







764



    code = TSDB_CODE_FAILED;




×







765



    goto _end;




×







766



  }










767












768



  if (cmpFn(key, &tmpKey) == 0) {





×







769



    *curKey = tmpKey;




×







770



    goto _end;




×







771



  } else if (!hasCurrentPrev) {





×







772



    code = TSDB_CODE_FAILED;




×







773



    goto _end;




×







774



  }










775












776



  sessionWinStateMoveToNext(pCur);




×







777



  code = sessionWinStateGetKVByCur(pCur, &tmpKey, NULL, NULL);




×







778



  if (code == TSDB_CODE_SUCCESS && cmpFn(key, &tmpKey) == 0) {





×







779



    *curKey = tmpKey;




×







780



  } else {










781



    code = TSDB_CODE_FAILED;




×







782



  }










783












784



_end:




×







785



  streamStateFreeCur(pCur);




×







786



  return code;




×







787



}










788












789



int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,




×







790



                              state_key_cmpr_fn fn, void** pVal, int32_t* pVLen, int32_t* pWinCode) {










791



  int32_t code = TSDB_CODE_SUCCESS;




×







792



  int32_t lino = 0;




×







793



  (*pWinCode) = TSDB_CODE_SUCCESS;




×







794












795



  SSessionKey* pWinKey = key;




×







796



  TSKEY        gap = 0;




×







797



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







798



  SArray*      pWinStates = NULL;




×







799












800



  void** ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







801



  if (ppBuff) {





×







802



    pWinStates = (SArray*)(*ppBuff);




×







803



  } else {










804



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







805



    if (!pWinStates) {





×







806



      code = terrno;




×







807



      QUERY_CHECK_CODE(code, lino, _end);





×







808



    }










809












810



    code = tSimpleHashPut(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);




×







811



    QUERY_CHECK_CODE(code, lino, _end);





×







812



  }










813












814



  TSKEY startTs = pWinKey->win.skey;




×







815



  TSKEY endTs = pWinKey->win.ekey;




×







816












817



  int32_t size = taosArrayGetSize(pWinStates);




×







818



  if (size == 0) {





×







819



    void*   pFileStore = getStateFileStore(pFileState);




×







820



    void*   p = NULL;




×







821



    int32_t code_file = streamStateStateAddIfNotExist_rocksdb(pFileStore, pWinKey, pKeyData, keyDataLen, fn, &p, pVLen);




×







822



    if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {





×







823



      (*pVal) = createSessionWinBuff(pFileState, pWinKey, p, pVLen);




×







824



      if (!(*pVal)) {





×







825



        code = TSDB_CODE_OUT_OF_MEMORY;




×







826



        QUERY_CHECK_CODE(code, lino, _end);





×







827



      }










828












829



      (*pWinCode) = code_file;




×







830



      qDebug("===stream===0 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,





×







831



             pWinKey->win.ekey, code_file);










832



    } else {










833



      code = addNewSessionWindow(pFileState, pWinStates, key, (SRowBuffPos**)pVal);




×







834



      (*pWinCode) = TSDB_CODE_FAILED;




×







835



      taosMemoryFree(p);





×







836



      QUERY_CHECK_CODE(code, lino, _end);





×







837



    }










838



    goto _end;




×







839



  }










840












841



  // find the first position which is smaller than the pWinKey










842



  int32_t      index = binarySearch(pWinStates, size, pWinKey, sessionStateKeyCompare);




×







843



  SRowBuffPos* pPos = NULL;




×







844



  int32_t      valSize = *pVLen;




×







845












846



  if (index >= 0) {





×







847



    pPos = taosArrayGetP(pWinStates, index);




×







848



    void* stateKey = (char*)(pPos->pRowBuff) + (valSize - keyDataLen);




×







849



    if (inSessionWindow(pPos->pKey, startTs, gap) || fn(pKeyData, stateKey) == true) {





×







850



      (*pVal) = pPos;




×







851



      SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;




×







852



      pPos->beUsed = true;




×







853



      pPos->beFlushed = false;




×







854



      *key = *pDestWinKey;




×







855



      goto _end;




×







856



    }










857



  }










858












859



  if (index + 1 < size) {





×







860



    pPos = taosArrayGetP(pWinStates, index + 1);




×







861



    void* stateKey = (char*)(pPos->pRowBuff) + (valSize - keyDataLen);




×







862



    if (inSessionWindow(pPos->pKey, startTs, gap) || (endTs != INT64_MIN && inSessionWindow(pPos->pKey, endTs, gap)) ||





×







863



        fn(pKeyData, stateKey) == true) {




×







864



      (*pVal) = pPos;




×







865



      SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;




×







866



      pPos->beUsed = true;




×







867



      pPos->beFlushed = false;




×







868



      *key = *pDestWinKey;




×







869



      goto _end;




×







870



    }










871



  }










872












873



  if (index + 1 == 0) {





×







874



    if (!isDeteled(pFileState, endTs)) {





×







875



      void*   p = NULL;




×







876



      void*   pFileStore = getStateFileStore(pFileState);




×







877



      int32_t code_file =










878



          streamStateStateAddIfNotExist_rocksdb(pFileStore, pWinKey, pKeyData, keyDataLen, fn, &p, pVLen);




×







879



      if (code_file == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {





×







880



        (*pVal) = createSessionWinBuff(pFileState, pWinKey, p, pVLen);




×







881



        if (!(*pVal)) {





×







882



          code = TSDB_CODE_OUT_OF_MEMORY;




×







883



          QUERY_CHECK_CODE(code, lino, _end);





×







884



        }










885












886



        (*pWinCode) = code_file;




×







887



        qDebug("===stream===1 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,





×







888



               pWinKey->win.ekey, code_file);










889



        goto _end;




×







890



      } else {










891



        taosMemoryFree(p);





×







892



      }










893



    }










894



  }










895












896



  if (index == size - 1) {





×







897



    code = addNewSessionWindow(pFileState, pWinStates, key, (SRowBuffPos**)pVal);




×







898



    QUERY_CHECK_CODE(code, lino, _end);





×







899












900



    (*pWinCode) = TSDB_CODE_FAILED;




×







901



    goto _end;




×







902



  }










903



  code = insertNewSessionWindow(pFileState, pWinStates, key, index + 1, (SRowBuffPos**)pVal);




×







904



  QUERY_CHECK_CODE(code, lino, _end);





×







905












906



  (*pWinCode) = TSDB_CODE_FAILED;




×







907












908



_end:




×







909



  if (code != TSDB_CODE_SUCCESS) {





×







910



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





×







911



  }










912



  return code;




×







913



}










914












915



int32_t getCountWinStateFromDisc(SStreamState* pState, SSessionKey* pKey, void** pVal, int32_t* pVLen) {




×







916



  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext_rocksdb(pState, pKey);




×







917



  int32_t          code = streamStateSessionGetKVByCur_rocksdb(pState, pCur, pKey, pVal, pVLen);




×







918



  streamStateFreeCur(pCur);




×







919



  if (code == TSDB_CODE_SUCCESS) {





×







920



    return code;




×







921



  } else {










922



    pCur = streamStateSessionSeekKeyPrev_rocksdb(pState, pKey);




×







923



  }










924












925



  code = streamStateSessionGetKVByCur_rocksdb(pState, pCur, pKey, pVal, pVLen);




×







926



  streamStateFreeCur(pCur);




×







927



  return code;




×







928



}










929












930



int32_t getCountWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, COUNT_TYPE winCount, void** pVal,




×







931



                              int32_t* pVLen, int32_t* pWinCount) {










932



  int32_t code = TSDB_CODE_SUCCESS;




×







933



  int32_t lino = 0;




×







934



  (*pWinCount) = TSDB_CODE_SUCCESS;




×







935












936



  SSessionKey* pWinKey = pKey;




×







937



  const TSKEY  gap = 0;




×







938



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







939



  SArray*      pWinStates = NULL;




×







940



  void**       ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







941



  if (ppBuff) {





×







942



    pWinStates = (SArray*)(*ppBuff);




×







943



  } else {










944



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







945



    if (!pWinStates) {





×







946



      code = terrno;




×







947



      QUERY_CHECK_CODE(code, lino, _end);





×







948



    }










949












950



    code = tSimpleHashPut(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);




×







951



    QUERY_CHECK_CODE(code, lino, _end);





×







952



  }










953












954



  TSKEY startTs = pWinKey->win.skey;




×







955



  TSKEY endTs = pWinKey->win.ekey;




×







956












957



  int32_t size = taosArrayGetSize(pWinStates);




×







958



  if (size == 0) {





×







959



    void* pFileStore = getStateFileStore(pFileState);




×







960



    void* pRockVal = NULL;




×







961



    (*pWinCount) = getCountWinStateFromDisc(pFileStore, pWinKey, &pRockVal, pVLen);




×







962



    if ((*pWinCount) == TSDB_CODE_SUCCESS || isFlushedState(pFileState, endTs, 0)) {





×







963



      qDebug("===stream===0 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,





×







964



             pWinKey->win.ekey, (*pWinCount));










965



      if ((*pWinCount) == TSDB_CODE_SUCCESS) {





×







966



        int32_t     valSize = *pVLen;




×







967



        COUNT_TYPE* pWinStateCout = (COUNT_TYPE*)((char*)(pRockVal) + (valSize - sizeof(COUNT_TYPE)));




×







968



        if (inSessionWindow(pWinKey, startTs, gap) || (*pWinStateCout) < winCount) {





×







969



          (*pVal) = createSessionWinBuff(pFileState, pWinKey, pRockVal, pVLen);




×







970



          if (!(*pVal)) {





×







971



            code = TSDB_CODE_OUT_OF_MEMORY;




×







972



            QUERY_CHECK_CODE(code, lino, _end);





×







973



          }










974












975



          goto _end;




×







976



        }










977



      }










978



      pWinKey->win.skey = startTs;




×







979



      pWinKey->win.ekey = endTs;




×







980



      (*pVal) = createSessionWinBuff(pFileState, pWinKey, NULL, NULL);




×







981



      taosMemoryFree(pRockVal);





×







982



      if (!(*pVal)) {





×







983



        code = TSDB_CODE_OUT_OF_MEMORY;




×







984



        QUERY_CHECK_CODE(code, lino, _end);





×







985



      }










986



    } else {










987



      code = addNewSessionWindow(pFileState, pWinStates, pWinKey, (SRowBuffPos**)pVal);




×







988



      QUERY_CHECK_CODE(code, lino, _end);





×







989












990



      (*pWinCount) = TSDB_CODE_FAILED;




×







991



    }










992



    goto _end;




×







993



  }










994












995



  // find the first position which is smaller than the pWinKey










996



  int32_t      index = binarySearch(pWinStates, size, pWinKey, sessionStateKeyCompare);




×







997



  SRowBuffPos* pPos = NULL;




×







998



  int32_t      valSize = *pVLen;




×







999












1000



  if (index >= 0) {





×







1001



    pPos = taosArrayGetP(pWinStates, index);




×







1002



    COUNT_TYPE* pWinStateCout = (COUNT_TYPE*)((char*)(pPos->pRowBuff) + (valSize - sizeof(COUNT_TYPE)));




×







1003



    if (inSessionWindow(pPos->pKey, startTs, gap) || (index == size - 1 && (*pWinStateCout) < winCount)) {





×







1004



      (*pVal) = pPos;




×







1005



      SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;




×







1006



      pPos->beUsed = true;




×







1007



      pPos->beFlushed = false;




×







1008



      *pWinKey = *pDestWinKey;




×







1009



      goto _end;




×







1010



    }










1011



  }










1012












1013



  if (index == -1) {





×







1014



    if (!isDeteled(pFileState, endTs) && isFlushedState(pFileState, endTs, 0)) {





×







1015



      SSessionKey tmpKey = *pWinKey;




×







1016



      void*       pRockVal = NULL;




×







1017



      void*       pFileStore = getStateFileStore(pFileState);




×







1018



      int32_t     code_file = getCountWinStateFromDisc(pFileStore, &tmpKey, &pRockVal, pVLen);




×







1019



      if (code_file == TSDB_CODE_SUCCESS) {





×







1020



        SRowBuffPos* pFirstPos = taosArrayGetP(pWinStates, 0);




×







1021



        SSessionKey* pFirstWinKey = (SSessionKey*)pFirstPos->pKey;




×







1022



        if (tmpKey.win.ekey < pFirstWinKey->win.skey) {





×







1023



          *pWinKey = tmpKey;




×







1024



          (*pVal) = createSessionWinBuff(pFileState, pWinKey, pRockVal, pVLen);




×







1025



          if (!(*pVal)) {





×







1026



            code = TSDB_CODE_OUT_OF_MEMORY;




×







1027



            QUERY_CHECK_CODE(code, lino, _end);





×







1028



          }










1029












1030



          (*pWinCount) = code_file;




×







1031



          qDebug("===stream===1 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,





×







1032



                 pWinKey->win.ekey, code_file);










1033



          goto _end;




×







1034



        }










1035



      }










1036



      taosMemoryFree(pRockVal);





×







1037



    }










1038



  }










1039












1040



  if (index + 1 < size) {





×







1041



    pPos = taosArrayGetP(pWinStates, index + 1);




×







1042



    (*pVal) = pPos;




×







1043



    SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;




×







1044



    pPos->beUsed = true;




×







1045



    pPos->beFlushed = false;




×







1046



    *pWinKey = *pDestWinKey;




×







1047



    goto _end;




×







1048



  }










1049












1050



  code = addNewSessionWindow(pFileState, pWinStates, pWinKey, (SRowBuffPos**)pVal);




×







1051



  QUERY_CHECK_CODE(code, lino, _end);





×







1052












1053



  (*pWinCount) = TSDB_CODE_FAILED;




×







1054












1055



_end:




×







1056



  if (code != TSDB_CODE_SUCCESS) {





×







1057



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





×







1058



  }










1059



  return code;




×







1060



}










1061












1062



int32_t createCountWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, COUNT_TYPE winCount, void** pVal,




×







1063



                                 int32_t* pVLen) {










1064



  SSessionKey* pWinKey = pKey;




×







1065



  const TSKEY  gap = 0;




×







1066



  int32_t      code = TSDB_CODE_SUCCESS;




×







1067



  int32_t      lino = 0;




×







1068



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







1069



  SArray*      pWinStates = NULL;




×







1070



  void**       ppBuff = tSimpleHashGet(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t));




×







1071



  if (ppBuff) {





×







1072



    pWinStates = (SArray*)(*ppBuff);




×







1073



  } else {










1074



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







1075



    code = tSimpleHashPut(pSessionBuff, &pWinKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);




×







1076



    QUERY_CHECK_CODE(code, lino, _end);





×







1077



  }










1078












1079



  TSKEY startTs = pWinKey->win.skey;




×







1080



  TSKEY endTs = pWinKey->win.ekey;




×







1081












1082



  int32_t size = taosArrayGetSize(pWinStates);




×







1083



  if (size == 0) {





×







1084



    void* pFileStore = getStateFileStore(pFileState);




×







1085



    void* pRockVal = NULL;




×







1086












1087



    int32_t code_file = getCountWinStateFromDisc(pFileStore, pWinKey, &pRockVal, pVLen);




×







1088



    if (code_file == TSDB_CODE_SUCCESS && isFlushedState(pFileState, endTs, 0)) {





×







1089



      int32_t     valSize = *pVLen;




×







1090



      COUNT_TYPE* pWinStateCount = (COUNT_TYPE*)((char*)(pRockVal) + (valSize - sizeof(COUNT_TYPE)));




×







1091



      if ((*pWinStateCount) == winCount) {





×







1092



        code = addNewSessionWindow(pFileState, pWinStates, pWinKey, (SRowBuffPos**)pVal);




×







1093



        QUERY_CHECK_CODE(code, lino, _end);





×







1094



      } else {










1095



        (*pVal) = createSessionWinBuff(pFileState, pWinKey, pRockVal, pVLen);




×







1096



        if (!(*pVal)) {





×







1097



          code = TSDB_CODE_OUT_OF_MEMORY;




×







1098



          QUERY_CHECK_CODE(code, lino, _end);





×







1099



        }










1100



        qDebug("===stream===0 get state win:%" PRId64 ",%" PRId64 " from disc, res %d", pWinKey->win.skey,





×







1101



               pWinKey->win.ekey, code_file);










1102



      }










1103



    } else {










1104



      code = addNewSessionWindow(pFileState, pWinStates, pWinKey, (SRowBuffPos**)pVal);




×







1105



      taosMemoryFree(pRockVal);





×







1106



      QUERY_CHECK_CODE(code, lino, _end);





×







1107



    }










1108



  } else {










1109



    code = addNewSessionWindow(pFileState, pWinStates, pWinKey, (SRowBuffPos**)pVal);




×







1110



    QUERY_CHECK_CODE(code, lino, _end);





×







1111



  }










1112












1113



_end:




×







1114



  if (code != TSDB_CODE_SUCCESS) {





×







1115



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





×







1116



  }










1117



  return code;




×







1118



}
























    

  • 

    •