taosdata / TDengine / #3633

coverage: 0.0% (-60.7%) from 60.719%
#3633

push

travis-ci

web-flow 
Merge pull request #30118 from taosdata/wl30

udpate ci workflow

0 of 280412 branches covered (0.0%)

Branch coverage included in aggregate %.

0 of 275582 relevant lines covered (0.0%)

0.0 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












566



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




×







567



  SArray*          pWinStates = NULL;




×







568



  int32_t          index = -1;




×







569



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




×







570



  if (pCur) {





×







571



    int32_t cmpRes= sessionStateRangeKeyCompare(pWinKey, pWinStates, index);




×







572



    if (cmpRes > 0) {





×







573



      return pCur;




×







574



    } else if (cmpRes == 0 && index > 0) {





×







575



      sessionWinStateMoveToPrev(pCur);




×







576



      return pCur;




×







577



    }










578



    streamStateFreeCur(pCur);




×







579



    pCur = NULL;




×







580



  }










581












582



  void* pFileStore = getStateFileStore(pFileState);




×







583



  pCur = streamStateSessionSeekKeyPrev_rocksdb(pFileStore, pWinKey);




×







584



  if (!pCur) {





×







585



    return NULL;




×







586



  }










587



  pCur->buffIndex = -1;




×







588



  pCur->pStreamFileState = pFileState;




×







589



  return pCur;




×







590



}










591












592



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




×







593



  if (!pCur) {





×







594



    return;




×







595



  }










596



  streamStateResetCur(pCur);




×







597



  pCur->buffIndex = 0;




×







598



  pCur->pStreamFileState = pFileState;




×







599



}










600












601



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




×







602



                                    SStreamStateCur** ppCur) {










603



  SSessionKey key = {.groupId = groupId};




×







604



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




×







605



  if (taosArrayGetSize(pWinStates) > 0 &&





×







606



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





×







607



    if (!(*ppCur)) {





×







608



      (*ppCur) = createStateCursor(pFileState);




×







609



    }










610



    transformCursor(pFileState, *ppCur);




×







611



  } else if (*ppCur) {





×







612



    (*ppCur)->buffIndex = -1;




×







613



    (*ppCur)->pStreamFileState = pFileState;




×







614



  }










615



}




×







616












617



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




×







618



  SArray*          pWinStates = NULL;




×







619



  int32_t          index = -1;




×







620



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




×







621



  if (pCur) {





×







622



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





×







623



      sessionWinStateMoveToNext(pCur);




×







624



    }










625



    return pCur;




×







626



  }










627












628



  void* pFileStore = getStateFileStore(pFileState);




×







629



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




×







630



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




×







631



  return pCur;




×







632



}










633












634



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




×







635



  SArray*          pWinStates = NULL;




×







636



  int32_t          index = -1;




×







637



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




×







638



  if (pCur) {





×







639



    sessionWinStateMoveToNext(pCur);




×







640



    return pCur;




×







641



  }










642












643



  void* pFileStore = getStateFileStore(pFileState);




×







644



  pCur = streamStateSessionSeekKeyNext_rocksdb(pFileStore, pWinKey);




×







645



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




×







646



  return pCur;




×







647



}










648












649



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




×







650



  SArray*          pWinStates = NULL;




×







651



  int32_t          index = -1;




×







652



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




×







653



  int32_t          resSize = getRowStateRowSize(pFileState);




×







654



  COUNT_TYPE       winCount = 0;




×







655



  if (pBuffCur) {





×







656



    while (index >= 0) {





×







657



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, index);




×







658



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




×







659



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





×







660



        index--;




×







661



      } else if (index >= 0) {





×







662



        pBuffCur->buffIndex = index + 1;




×







663



        return pBuffCur;




×







664



      }










665



    }










666



    pBuffCur->buffIndex = 0;




×







667



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





×







668



    pBuffCur = createStateCursor(pFileState);




×







669



    if (pBuffCur == NULL) {





×







670



      return NULL;




×







671



    }










672



    pBuffCur->buffIndex = 0;




×







673



  }










674












675



  void*            pFileStore = getStateFileStore(pFileState);




×







676



  SStreamStateCur* pCur = streamStateSessionSeekKeyPrev_rocksdb(pFileStore, pWinKey);




×







677



  if (pCur) {





×







678



    pCur->pStreamFileState = pFileState;




×







679



    SSessionKey key = {0};




×







680



    void*       pVal = NULL;




×







681



    int         len = 0;




×







682



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




×







683



    if (code == TSDB_CODE_FAILED) {





×







684



      streamStateFreeCur(pCur);




×







685



      return pBuffCur;




×







686



    }










687



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




×







688



    taosMemoryFreeClear(pVal);





×







689



    streamStateFreeCur(pBuffCur);




×







690



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





×







691



      streamStateCurNext(pFileStore, pCur);




×







692



      return pCur;




×







693



    }










694



    streamStateCurPrev(pFileStore, pCur);




×







695



    while (1) {










696



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




×







697



      if (code == TSDB_CODE_FAILED) {





×







698



        streamStateCurNext(pFileStore, pCur);




×







699



        return pCur;




×







700



      }










701



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




×







702



      taosMemoryFreeClear(pVal);





×







703



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





×







704



        streamStateCurPrev(pFileStore, pCur);




×







705



      } else {










706



        streamStateCurNext(pFileStore, pCur);




×







707



        return pCur;




×







708



      }










709



    }










710



  }










711



  return pBuffCur;




×







712



}










713












714



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




×







715



  if (!pCur) {





×







716



    return TSDB_CODE_FAILED;




×







717



  }










718



  int32_t code = TSDB_CODE_SUCCESS;




×







719












720



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




×







721



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




×







722



  SArray*    pWinStates = NULL;




×







723



  if (ppBuff) {





×







724



    pWinStates = (SArray*)(*ppBuff);




×







725



  }










726












727



  if (pCur->buffIndex >= 0) {





×







728



    int32_t size = taosArrayGetSize(pWinStates);




×







729



    if (pCur->buffIndex >= size) {





×







730



      return TSDB_CODE_FAILED;




×







731



    }










732



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




×







733



    if (pVal) {





×







734



      *pVal = pPos;




×







735



    }










736



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




×







737



  } else {










738



    void* pData = NULL;




×







739



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




×







740



    if (taosArrayGetSize(pWinStates) > 0 &&





×







741



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





×







742



      transformCursor(pCur->pStreamFileState, pCur);




×







743



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




×







744



      if (pVal) {





×







745



        *pVal = pPos;




×







746



      }










747



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




×







748



      code = TSDB_CODE_SUCCESS;




×







749



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





×







750



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




×







751



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





×







752



        code = TSDB_CODE_OUT_OF_MEMORY;




×







753



        taosMemoryFreeClear(pData);





×







754



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





×







755



        return code;




×







756



      }










757



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




×







758



      pNewPos->needFree = true;




×







759



      pNewPos->beFlushed = true;




×







760



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




×







761



      (*pVal) = pNewPos;




×







762



    }










763



    taosMemoryFreeClear(pData);





×







764



  }










765



  return code;




×







766



}










767












768



void sessionWinStateMoveToNext(SStreamStateCur* pCur) {




×







769



  qTrace("move cursor to next");





×







770



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





×







771



    pCur->buffIndex++;




×







772



  } else {










773



    streamStateCurNext_rocksdb(pCur);




×







774



  }










775



}




×







776












777



void sessionWinStateMoveToPrev(SStreamStateCur* pCur) {




×







778



  qTrace("move cursor to prev");





×







779



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





×







780



    pCur->buffIndex--;




×







781



  } else {










782



    streamStateCurPrev_rocksdb(pCur);




×







783



  }










784



}




×







785












786



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




×







787



                                     range_cmpr_fn cmpFn) {










788



  SStreamStateCur* pCur = sessionWinStateSeekKeyCurrentPrev(pFileState, key);




×







789



  SSessionKey      tmpKey = *key;




×







790



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




×







791



  bool             hasCurrentPrev = true;




×







792



  if (code == TSDB_CODE_FAILED) {





×







793



    streamStateFreeCur(pCur);




×







794



    pCur = sessionWinStateSeekKeyNext(pFileState, key);




×







795



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




×







796



    hasCurrentPrev = false;




×







797



  }










798












799



  if (code == TSDB_CODE_FAILED) {





×







800



    code = TSDB_CODE_FAILED;




×







801



    goto _end;




×







802



  }










803












804



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





×







805



    *curKey = tmpKey;




×







806



    goto _end;




×







807



  } else if (!hasCurrentPrev) {





×







808



    code = TSDB_CODE_FAILED;




×







809



    goto _end;




×







810



  }










811












812



  sessionWinStateMoveToNext(pCur);




×







813



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




×







814



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





×







815



    *curKey = tmpKey;




×







816



  } else {










817



    code = TSDB_CODE_FAILED;




×







818



  }










819












820



_end:




×







821



  streamStateFreeCur(pCur);




×







822



  return code;




×







823



}










824












825



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




×







826



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










827



  int32_t code = TSDB_CODE_SUCCESS;




×







828



  int32_t lino = 0;




×







829



  (*pWinCode) = TSDB_CODE_SUCCESS;




×







830












831



  SSessionKey* pWinKey = key;




×







832



  TSKEY        gap = 0;




×







833



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







834



  SArray*      pWinStates = NULL;




×







835












836



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




×







837



  if (ppBuff) {





×







838



    pWinStates = (SArray*)(*ppBuff);




×







839



  } else {










840



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







841



    if (!pWinStates) {





×







842



      code = terrno;




×







843



      QUERY_CHECK_CODE(code, lino, _end);





×







844



    }










845












846



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




×







847



    QUERY_CHECK_CODE(code, lino, _end);





×







848



  }










849












850



  TSKEY startTs = pWinKey->win.skey;




×







851



  TSKEY endTs = pWinKey->win.ekey;




×







852












853



  int32_t size = taosArrayGetSize(pWinStates);




×







854



  if (size == 0) {





×







855



    void*   pFileStore = getStateFileStore(pFileState);




×







856



    void*   p = NULL;




×







857



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




×







858



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





×







859



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




×







860



      if (!(*pVal)) {





×







861



        code = TSDB_CODE_OUT_OF_MEMORY;




×







862



        QUERY_CHECK_CODE(code, lino, _end);





×







863



      }










864












865



      (*pWinCode) = code_file;




×







866



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





×







867



             pWinKey->win.ekey, code_file);










868



    } else {










869



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




×







870



      (*pWinCode) = TSDB_CODE_FAILED;




×







871



      taosMemoryFree(p);





×







872



      QUERY_CHECK_CODE(code, lino, _end);





×







873



    }










874



    goto _end;




×







875



  }










876












877



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










878



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




×







879



  SRowBuffPos* pPos = NULL;




×







880



  int32_t      valSize = *pVLen;




×







881












882



  if (index >= 0) {





×







883



    pPos = taosArrayGetP(pWinStates, index);




×







884



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




×







885



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





×







886



      (*pVal) = pPos;




×







887



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




×







888



      pPos->beUsed = true;




×







889



      pPos->beFlushed = false;




×







890



      *key = *pDestWinKey;




×







891



      goto _end;




×







892



    }










893



  }










894












895



  if (index + 1 < size) {





×







896



    pPos = taosArrayGetP(pWinStates, index + 1);




×







897



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




×







898



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





×







899



        fn(pKeyData, stateKey) == true) {




×







900



      (*pVal) = pPos;




×







901



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




×







902



      pPos->beUsed = true;




×







903



      pPos->beFlushed = false;




×







904



      *key = *pDestWinKey;




×







905



      goto _end;




×







906



    }










907



  }










908












909



  if (index + 1 == 0) {





×







910



    if (!isDeteled(pFileState, endTs)) {





×







911



      void*   p = NULL;




×







912



      void*   pFileStore = getStateFileStore(pFileState);




×







913



      int32_t code_file =










914



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




×







915



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





×







916



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




×







917



        if (!(*pVal)) {





×







918



          code = TSDB_CODE_OUT_OF_MEMORY;




×







919



          QUERY_CHECK_CODE(code, lino, _end);





×







920



        }










921












922



        (*pWinCode) = code_file;




×







923



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





×







924



               pWinKey->win.ekey, code_file);










925



        goto _end;




×







926



      } else {










927



        taosMemoryFree(p);





×







928



      }










929



    }










930



  }










931












932



  if (index == size - 1) {





×







933



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




×







934



    QUERY_CHECK_CODE(code, lino, _end);





×







935












936



    (*pWinCode) = TSDB_CODE_FAILED;




×







937



    goto _end;




×







938



  }










939



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




×







940



  QUERY_CHECK_CODE(code, lino, _end);





×







941












942



  (*pWinCode) = TSDB_CODE_FAILED;




×







943












944



_end:




×







945



  if (code != TSDB_CODE_SUCCESS) {





×







946



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





×







947



  }










948



  return code;




×







949



}










950












951



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




×







952



  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext_rocksdb(pState, pKey);




×







953



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




×







954



  streamStateFreeCur(pCur);




×







955



  if (code == TSDB_CODE_SUCCESS) {





×







956



    return code;




×







957



  } else {










958



    pCur = streamStateSessionSeekKeyPrev_rocksdb(pState, pKey);




×







959



  }










960












961



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




×







962



  streamStateFreeCur(pCur);




×







963



  return code;




×







964



}










965












966



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




×







967



                              int32_t* pVLen, int32_t* pWinCount) {










968



  int32_t code = TSDB_CODE_SUCCESS;




×







969



  int32_t lino = 0;




×







970



  (*pWinCount) = TSDB_CODE_SUCCESS;




×







971












972



  SSessionKey* pWinKey = pKey;




×







973



  const TSKEY  gap = 0;




×







974



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







975



  SArray*      pWinStates = NULL;




×







976



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




×







977



  if (ppBuff) {





×







978



    pWinStates = (SArray*)(*ppBuff);




×







979



  } else {










980



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







981



    if (!pWinStates) {





×







982



      code = terrno;




×







983



      QUERY_CHECK_CODE(code, lino, _end);





×







984



    }










985












986



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




×







987



    QUERY_CHECK_CODE(code, lino, _end);





×







988



  }










989












990



  TSKEY startTs = pWinKey->win.skey;




×







991



  TSKEY endTs = pWinKey->win.ekey;




×







992












993



  int32_t size = taosArrayGetSize(pWinStates);




×







994



  if (size == 0) {





×







995



    void* pFileStore = getStateFileStore(pFileState);




×







996



    void* pRockVal = NULL;




×







997



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




×







998



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





×







999



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





×







1000



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










1001



      if ((*pWinCount) == TSDB_CODE_SUCCESS) {





×







1002



        int32_t     valSize = *pVLen;




×







1003



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




×







1004



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





×







1005



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




×







1006



          if (!(*pVal)) {





×







1007



            code = TSDB_CODE_OUT_OF_MEMORY;




×







1008



            QUERY_CHECK_CODE(code, lino, _end);





×







1009



          }










1010












1011



          goto _end;




×







1012



        }










1013



      }










1014



      pWinKey->win.skey = startTs;




×







1015



      pWinKey->win.ekey = endTs;




×







1016



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




×







1017



      taosMemoryFree(pRockVal);





×







1018



      if (!(*pVal)) {





×







1019



        code = TSDB_CODE_OUT_OF_MEMORY;




×







1020



        QUERY_CHECK_CODE(code, lino, _end);





×







1021



      }










1022



    } else {










1023



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




×







1024



      QUERY_CHECK_CODE(code, lino, _end);





×







1025












1026



      (*pWinCount) = TSDB_CODE_FAILED;




×







1027



    }










1028



    goto _end;




×







1029



  }










1030












1031



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










1032



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




×







1033



  SRowBuffPos* pPos = NULL;




×







1034



  int32_t      valSize = *pVLen;




×







1035












1036



  if (index >= 0) {





×







1037



    pPos = taosArrayGetP(pWinStates, index);




×







1038



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




×







1039



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





×







1040



      (*pVal) = pPos;




×







1041



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




×







1042



      pPos->beUsed = true;




×







1043



      pPos->beFlushed = false;




×







1044



      *pWinKey = *pDestWinKey;




×







1045



      goto _end;




×







1046



    }










1047



  }










1048












1049



  if (index == -1) {





×







1050



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





×







1051



      SSessionKey tmpKey = *pWinKey;




×







1052



      void*       pRockVal = NULL;




×







1053



      void*       pFileStore = getStateFileStore(pFileState);




×







1054



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




×







1055



      if (code_file == TSDB_CODE_SUCCESS) {





×







1056



        SRowBuffPos* pFirstPos = taosArrayGetP(pWinStates, 0);




×







1057



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




×







1058



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





×







1059



          *pWinKey = tmpKey;




×







1060



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




×







1061



          if (!(*pVal)) {





×







1062



            code = TSDB_CODE_OUT_OF_MEMORY;




×







1063



            QUERY_CHECK_CODE(code, lino, _end);





×







1064



          }










1065












1066



          (*pWinCount) = code_file;




×







1067



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





×







1068



                 pWinKey->win.ekey, code_file);










1069



          goto _end;




×







1070



        }










1071



      }










1072



      taosMemoryFree(pRockVal);





×







1073



    }










1074



  }










1075












1076



  if (index + 1 < size) {





×







1077



    pPos = taosArrayGetP(pWinStates, index + 1);




×







1078



    (*pVal) = pPos;




×







1079



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




×







1080



    pPos->beUsed = true;




×







1081



    pPos->beFlushed = false;




×







1082



    *pWinKey = *pDestWinKey;




×







1083



    goto _end;




×







1084



  }










1085












1086



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




×







1087



  QUERY_CHECK_CODE(code, lino, _end);





×







1088












1089



  (*pWinCount) = TSDB_CODE_FAILED;




×







1090












1091



_end:




×







1092



  if (code != TSDB_CODE_SUCCESS) {





×







1093



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





×







1094



  }










1095



  return code;




×







1096



}










1097












1098



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




×







1099



                                 int32_t* pVLen) {










1100



  SSessionKey* pWinKey = pKey;




×







1101



  const TSKEY  gap = 0;




×







1102



  int32_t      code = TSDB_CODE_SUCCESS;




×







1103



  int32_t      lino = 0;




×







1104



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×







1105



  SArray*      pWinStates = NULL;




×







1106



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




×







1107



  if (ppBuff) {





×







1108



    pWinStates = (SArray*)(*ppBuff);




×







1109



  } else {










1110



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×







1111



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




×







1112



    QUERY_CHECK_CODE(code, lino, _end);





×







1113



  }










1114












1115



  TSKEY startTs = pWinKey->win.skey;




×







1116



  TSKEY endTs = pWinKey->win.ekey;




×







1117












1118



  int32_t size = taosArrayGetSize(pWinStates);




×







1119



  if (size == 0) {





×







1120



    void* pFileStore = getStateFileStore(pFileState);




×







1121



    void* pRockVal = NULL;




×







1122












1123



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




×







1124



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





×







1125



      int32_t     valSize = *pVLen;




×







1126



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




×







1127



      if ((*pWinStateCount) == winCount) {





×







1128



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




×







1129



        QUERY_CHECK_CODE(code, lino, _end);





×







1130



      } else {










1131



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




×







1132



        if (!(*pVal)) {





×







1133



          code = TSDB_CODE_OUT_OF_MEMORY;




×







1134



          QUERY_CHECK_CODE(code, lino, _end);





×







1135



        }










1136



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





×







1137



               pWinKey->win.ekey, code_file);










1138



      }










1139



    } else {










1140



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




×







1141



      taosMemoryFree(pRockVal);





×







1142



      QUERY_CHECK_CODE(code, lino, _end);





×







1143



    }










1144



  } else {










1145



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




×







1146



    QUERY_CHECK_CODE(code, lino, _end);





×







1147



  }










1148












1149



_end:




×







1150



  if (code != TSDB_CODE_SUCCESS) {





×







1151



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





×







1152



  }










1153



  return code;




×







1154



}
























    

  • 

    •