taosdata / TDengine / #3621

coverage: 2.037% (-61.5%) from 63.573%
#3621

push

travis-ci

web-flow 
Merge pull request #29874 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 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













UNCOV

23



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




×








UNCOV

24



  SRowBuffPos* pPos2 = taosArrayGetP(pDatas, pos);




×








UNCOV

25



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




×








UNCOV

26



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




×







27



}










28













UNCOV

29



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




×








UNCOV

30



  SRowBuffPos* pPos2 = taosArrayGetP(pDatas, pos);




×








UNCOV

31



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




×








UNCOV

32



  return sessionRangeKeyCmpr(pWin1, pWin2);




×







33



}










34













UNCOV

35



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




×








UNCOV

36



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




×








UNCOV

37



  int numOfRows = 0;




×







38













UNCOV

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) {











UNCOV

43



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





×








UNCOV

44



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





×








UNCOV

45



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





×







46













UNCOV

47



    numOfRows = lastPos - firstPos + 1;




×








UNCOV

48



    midPos = (numOfRows >> 1) + firstPos;




×







49













UNCOV

50



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





×








UNCOV

51



      lastPos = midPos - 1;




×








UNCOV

52



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





×








UNCOV

53



      firstPos = midPos + 1;




×







54



    } else {











UNCOV

55



      break;




×







56



    }










57



  }










58













UNCOV

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













UNCOV

73



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




×








UNCOV

74



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





×








UNCOV

75



    return true;




×







76



  }











UNCOV

77



  return false;




×







78



}










79













UNCOV

80



SStreamStateCur* createStateCursor(SStreamFileState* pFileState) {




×








UNCOV

81



  SStreamStateCur* pCur = createStreamStateCursor();




×








UNCOV

82



  if (pCur == NULL) {





×







83



    return NULL;




×







84



  }











UNCOV

85



  pCur->pStreamFileState = pFileState;




×








UNCOV

86



  return pCur;




×







87



}










88













UNCOV

89



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




×







90



                                   SRowBuffPos** ppPos) {











UNCOV

91



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

92



  int32_t      lino = 0;




×








UNCOV

93



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×








UNCOV

94



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





×







95



    code = TSDB_CODE_OUT_OF_MEMORY;




×







96



    QUERY_CHECK_CODE(code, lino, _end);





×







97



  }










98













UNCOV

99



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




×








UNCOV

100



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




×








UNCOV

101



  if (!tmp) {





×







102



    code = terrno;




×







103



    QUERY_CHECK_CODE(code, lino, _end);





×







104



  }











UNCOV

105



  (*ppPos) = pNewPos;




×







106













UNCOV

107



_end:




×








UNCOV

108



  if (code != TSDB_CODE_SUCCESS) {





×







109



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





×







110



  }











UNCOV

111



  return code;




×







112



}










113













UNCOV

114



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




×







115



                                      int32_t index, SRowBuffPos** ppPos) {











UNCOV

116



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

117



  int32_t      lino = 0;




×








UNCOV

118



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×








UNCOV

119



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





×







120



    code = TSDB_CODE_OUT_OF_MEMORY;




×







121



    QUERY_CHECK_CODE(code, lino, _end);





×







122



  }










123













UNCOV

124



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




×








UNCOV

125



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




×








UNCOV

126



  if (!tmp) {





×







127



    code = terrno;




×







128



    QUERY_CHECK_CODE(code, lino, _end);





×







129



  }










130













UNCOV

131



  *ppPos = pNewPos;




×







132













UNCOV

133



_end:




×








UNCOV

134



  if (code != TSDB_CODE_SUCCESS) {





×







135



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





×







136



  }











UNCOV

137



  return code;




×







138



}










139













UNCOV

140



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




×








UNCOV

141



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

142



  int32_t      lino = 0;




×








UNCOV

143



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×








UNCOV

144



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





×







145



    code = TSDB_CODE_OUT_OF_MEMORY;




×







146



    QUERY_CHECK_CODE(code, lino, _end);





×







147



  }










148













UNCOV

149



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




×








UNCOV

150



  pNewPos->needFree = true;




×








UNCOV

151



  pNewPos->beFlushed = true;




×








UNCOV

152



  if (p) {





×








UNCOV

153



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




×







154



  } else {











UNCOV

155



    int32_t len = getRowStateRowSize(pFileState);




×








UNCOV

156



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




×







157



  }










158













UNCOV

159



_end:




×








UNCOV

160



  taosMemoryFree(p);





×








UNCOV

161



  if (code != TSDB_CODE_SUCCESS) {





×







162



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





×







163



    return NULL;




×







164



  }











UNCOV

165



  return pNewPos;




×







166



}










167













UNCOV

168



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




×







169



                                int32_t* pWinCode) {











UNCOV

170



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

171



  int32_t lino = 0;




×








UNCOV

172



  (*pWinCode) = TSDB_CODE_SUCCESS;




×








UNCOV

173



  SSHashObj* pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

174



  SArray*    pWinStates = NULL;




×








UNCOV

175



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




×








UNCOV

176



  if (ppBuff) {





×








UNCOV

177



    pWinStates = (SArray*)(*ppBuff);




×







178



  } else {











UNCOV

179



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×








UNCOV

180



    if (!pWinStates) {





×







181



      code = terrno;




×







182



      QUERY_CHECK_CODE(code, lino, _end);





×







183



    }











UNCOV

184



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




×








UNCOV

185



    QUERY_CHECK_CODE(code, lino, _end);





×







186



  }










187













UNCOV

188



  TSKEY startTs = pKey->win.skey;




×








UNCOV

189



  TSKEY endTs = pKey->win.ekey;




×







190













UNCOV

191



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

192



  if (size == 0) {





×








UNCOV

193



    void*   pFileStore = getStateFileStore(pFileState);




×








UNCOV

194



    void*   p = NULL;




×








UNCOV

195



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




×








UNCOV

196



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





×








UNCOV

197



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




×








UNCOV

198



      if (!(*pVal)) {





×







199



        code = TSDB_CODE_OUT_OF_MEMORY;




×







200



        QUERY_CHECK_CODE(code, lino, _end);





×







201



      }










202













UNCOV

203



      (*pWinCode) = code_file;




×








UNCOV

204



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





×







205



    } else {











UNCOV

206



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




×








UNCOV

207



      (*pWinCode) = TSDB_CODE_FAILED;




×








UNCOV

208



      taosMemoryFree(p);





×








UNCOV

209



      QUERY_CHECK_CODE(code, lino, _end);





×







210



    }











UNCOV

211



    goto _end;




×







212



  }










213












214



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











UNCOV

215



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




×








UNCOV

216



  SRowBuffPos* pPos = NULL;




×







217













UNCOV

218



  if (index >= 0) {





×








UNCOV

219



    pPos = taosArrayGetP(pWinStates, index);




×








UNCOV

220



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





×








UNCOV

221



      (*pVal) = pPos;




×








UNCOV

222



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




×








UNCOV

223



      pPos->beUsed = true;




×








UNCOV

224



      pPos->beFlushed = false;




×








UNCOV

225



      *pKey = *pDestWinKey;




×








UNCOV

226



      goto _end;




×







227



    }










228



  }










229













UNCOV

230



  if (index + 1 < size) {





×








UNCOV

231



    pPos = taosArrayGetP(pWinStates, index + 1);




×








UNCOV

232



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





×








UNCOV

233



      (*pVal) = pPos;




×








UNCOV

234



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




×








UNCOV

235



      pPos->beUsed = true;




×








UNCOV

236



      pPos->beFlushed = false;




×








UNCOV

237



      *pKey = *pDestWinKey;




×








UNCOV

238



      goto _end;




×







239



    }










240



  }










241













UNCOV

242



  if (index + 1 == 0) {





×








UNCOV

243



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





×








UNCOV

244



      void*   p = NULL;




×








UNCOV

245



      void*   pFileStore = getStateFileStore(pFileState);




×








UNCOV

246



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




×








UNCOV

247



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





×








UNCOV

248



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




×








UNCOV

249



        if (!(*pVal)) {





×







250



          code = TSDB_CODE_OUT_OF_MEMORY;




×








UNCOV

251



          QUERY_CHECK_CODE(code, lino, _end);





×







252



        }










253













UNCOV

254



        (*pWinCode) = code_file;




×








UNCOV

255



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





×








UNCOV

256



        goto _end;




×







257



      } else {










258



        taosMemoryFree(p);





×







259



      }










260



    }










261



  }










262













UNCOV

263



  if (index == size - 1) {





×








UNCOV

264



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




×








UNCOV

265



    QUERY_CHECK_CODE(code, lino, _end);





×







266













UNCOV

267



    (*pWinCode) = TSDB_CODE_FAILED;




×








UNCOV

268



    goto _end;




×







269



  }










270













UNCOV

271



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




×








UNCOV

272



  QUERY_CHECK_CODE(code, lino, _end);





×







273













UNCOV

274



  (*pWinCode) = TSDB_CODE_FAILED;




×







275













UNCOV

276



_end:




×








UNCOV

277



  if (code != TSDB_CODE_SUCCESS) {





×







278



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





×







279



  }











UNCOV

280



  return code;




×







281



}










282













UNCOV

283



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




×







284



                          int32_t* pWinCode) {











UNCOV

285



  SWinKey*    pTmpkey = pKey;




×








UNCOV

286



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




×








UNCOV

287



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




×







288



}










289













UNCOV

290



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




×








UNCOV

291



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

292



  int32_t      lino = 0;




×








UNCOV

293



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

294



  SSessionKey* pKey = pPos->pKey;




×








UNCOV

295



  SArray*      pWinStates = NULL;




×








UNCOV

296



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




×








UNCOV

297



  if (ppBuff) {





×








UNCOV

298



    pWinStates = (SArray*)(*ppBuff);




×







299



  } else {











UNCOV

300



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×








UNCOV

301



    if (!pWinStates) {





×







302



      code = terrno;




×







303



      QUERY_CHECK_CODE(code, lino, _end);





×







304



    }










305













UNCOV

306



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




×








UNCOV

307



    QUERY_CHECK_CODE(code, lino, _end);





×







308



  }










309













UNCOV

310



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

311



  if (size == 0) {





×








UNCOV

312



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




×








UNCOV

313



    if (!tmp) {





×







314



      code = terrno;




×







315



      QUERY_CHECK_CODE(code, lino, _end);





×







316



    }











UNCOV

317



    goto _end;




×







318



  }










319












320



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











UNCOV

321



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




×








UNCOV

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 {











UNCOV

329



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




×








UNCOV

330



    if (!tmp) {





×







331



      code = terrno;




×







332



      QUERY_CHECK_CODE(code, lino, _end);





×







333



    }










334



  }










335













UNCOV

336



_end:




×








UNCOV

337



  pPos->needFree = false;




×








UNCOV

338



  if (code != TSDB_CODE_SUCCESS) {





×







339



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





×







340



  }











UNCOV

341



  return code;




×







342



}










343













UNCOV

344



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




×







345



                              int32_t* pWinCode) {











UNCOV

346



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

347



  int32_t      lino = 0;




×








UNCOV

348



  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);




×








UNCOV

349



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





×







350



    code = TSDB_CODE_OUT_OF_MEMORY;




×







351



    QUERY_CHECK_CODE(code, lino, _end);





×







352



  }











UNCOV

353



  pNewPos->needFree = true;




×








UNCOV

354



  pNewPos->beFlushed = true;




×








UNCOV

355



  void* pBuff = NULL;




×








UNCOV

356



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




×








UNCOV

357



  if ((*pWinCode) != TSDB_CODE_SUCCESS) {





×







358



    goto _end;




×







359



  }











UNCOV

360



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




×








UNCOV

361



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




×








UNCOV

362



  taosMemoryFreeClear(pBuff);





×








UNCOV

363



  (*pVal) = pNewPos;




×







364













UNCOV

365



_end:




×








UNCOV

366



  if (code != TSDB_CODE_SUCCESS) {





×







367



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





×







368



  }











UNCOV

369



  return code;




×







370



}










371













UNCOV

372



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




×








UNCOV

373



  SSHashObj*   pSessionBuff = (SSHashObj*)pBuff;




×








UNCOV

374



  SSessionKey* pWinKey = (SSessionKey*)key;




×








UNCOV

375



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




×








UNCOV

376



  if (!ppBuff) {





×








UNCOV

377



    return TSDB_CODE_SUCCESS;




×







378



  }











UNCOV

379



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




×








UNCOV

380



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

381



  TSKEY   gap = 0;




×








UNCOV

382



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




×








UNCOV

383



  if (index >= 0) {





×








UNCOV

384



    SRowBuffPos* pPos = taosArrayGetP(pWinStates, index);




×








UNCOV

385



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





×








UNCOV

386



      pPos->beFlushed = true;




×








UNCOV

387



      taosArrayRemove(pWinStates, index);




×







388



    }










389



  }











UNCOV

390



  return TSDB_CODE_SUCCESS;




×







391



}










392













UNCOV

393



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




×








UNCOV

394



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

395



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




×








UNCOV

396



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




×








UNCOV

397



  if (!ppBuff) {





×







398



    return;




×







399



  }











UNCOV

400



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




×








UNCOV

401



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

402



  TSKEY   gap = 0;




×








UNCOV

403



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




×








UNCOV

404



  if (index >= 0) {





×








UNCOV

405



    SRowBuffPos* pItemPos = taosArrayGetP(pWinStates, index);




×








UNCOV

406



    if (pItemPos == pPos) {





×








UNCOV

407



      taosArrayRemove(pWinStates, index);




×







408



    }










409



  }










410



}










411













UNCOV

412



int32_t allocSessioncWinBuffByNextPosition(SStreamFileState* pFileState, SStreamStateCur* pCur,




×







413



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











UNCOV

414



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

415



  int32_t      lino = 0;




×








UNCOV

416



  SRowBuffPos* pNewPos = NULL;




×








UNCOV

417



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

418



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




×








UNCOV

419



  SArray*      pWinStates = NULL;




×








UNCOV

420



  if (!ppBuff) {





×








UNCOV

421



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×








UNCOV

422



    if (!pWinStates) {





×







423



      code = terrno;




×







424



      QUERY_CHECK_CODE(code, lino, _end);





×







425



    }










426













UNCOV

427



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




×








UNCOV

428



    QUERY_CHECK_CODE(code, lino, _end);





×







429



  } else {











UNCOV

430



    pWinStates = (SArray*)(*ppBuff);




×







431



  }











UNCOV

432



  if (!pCur) {





×








UNCOV

433



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




×








UNCOV

434



    QUERY_CHECK_CODE(code, lino, _end);





×







435













UNCOV

436



    goto _end;




×







437



  }










438













UNCOV

439



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

440



  if (pCur->buffIndex >= 0) {





×








UNCOV

441



    if (pCur->buffIndex >= size) {





×








UNCOV

442



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




×








UNCOV

443



      QUERY_CHECK_CODE(code, lino, _end);





×







444













UNCOV

445



      goto _end;




×







446



    }











UNCOV

447



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




×








UNCOV

448



    QUERY_CHECK_CODE(code, lino, _end);





×







449













UNCOV

450



    goto _end;




×







451



  } else {











UNCOV

452



    if (size > 0) {





×








UNCOV

453



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, 0);




×








UNCOV

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



    }











UNCOV

464



    pNewPos = getNewRowPosForWrite(pFileState);




×








UNCOV

465



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





×







466



      code = TSDB_CODE_OUT_OF_MEMORY;




×







467



      QUERY_CHECK_CODE(code, lino, _end);





×







468



    }










469













UNCOV

470



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




×








UNCOV

471



    pNewPos->needFree = true;




×








UNCOV

472



    pNewPos->beFlushed = true;




×







473



  }










474













UNCOV

475



_end:




×








UNCOV

476



  (*ppVal) = pNewPos;




×








UNCOV

477



  if (code != TSDB_CODE_SUCCESS) {





×







478



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





×







479



  }











UNCOV

480



  return code;




×







481



}










482













UNCOV

483



void sessionWinStateClear(SStreamFileState* pFileState) {




×








UNCOV

484



  int32_t buffSize = getRowStateRowSize(pFileState);




×








UNCOV

485



  void*   pIte = NULL;




×








UNCOV

486



  size_t  keyLen = 0;




×








UNCOV

487



  int32_t iter = 0;




×








UNCOV

488



  void*   pBuff = getRowStateBuff(pFileState);




×








UNCOV

489



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





×








UNCOV

490



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




×








UNCOV

491



    int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

492



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





×








UNCOV

493



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, i);




×








UNCOV

494



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




×







495



    }










496



  }











UNCOV

497



}




×







498













UNCOV

499



void sessionWinStateCleanup(void* pBuff) {




×








UNCOV

500



  void*   pIte = NULL;




×








UNCOV

501



  size_t  keyLen = 0;




×








UNCOV

502



  int32_t iter = 0;




×








UNCOV

503



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





×








UNCOV

504



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




×








UNCOV

505



    taosArrayDestroy(pWinStates);




×







506



  }











UNCOV

507



  tSimpleHashCleanup(pBuff);




×








UNCOV

508



}




×







509













UNCOV

510



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




×







511



                                                SArray** pWins, int32_t* pIndex) {











UNCOV

512



  SStreamStateCur* pCur = NULL;




×








UNCOV

513



  SSHashObj*       pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

514



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




×








UNCOV

515



  if (!ppBuff) {





×








UNCOV

516



    return NULL;




×







517



  }










518













UNCOV

519



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




×








UNCOV

520



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

521



  TSKEY   gap = 0;




×








UNCOV

522



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




×







523













UNCOV

524



  if (pWins) {





×








UNCOV

525



    (*pWins) = pWinStates;




×







526



  }










527













UNCOV

528



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





×








UNCOV

529



    SRowBuffPos* pPos = taosArrayGetP(pWinStates, 0);




×








UNCOV

530



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




×








UNCOV

531



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





×








UNCOV

532



      index = 0;




×







533



    }










534



  }










535













UNCOV

536



  if (index >= 0) {





×








UNCOV

537



    pCur = createStateCursor(pFileState);




×








UNCOV

538



    if (pCur == NULL) {





×







539



      return NULL;




×







540



    }











UNCOV

541



    pCur->buffIndex = index;




×








UNCOV

542



    if (pIndex) {





×








UNCOV

543



      *pIndex = index;




×







544



    }










545



  }










546













UNCOV

547



  return pCur;




×







548



}










549













UNCOV

550



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




×








UNCOV

551



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




×








UNCOV

552



  if (pCur) {





×








UNCOV

553



    return pCur;




×







554



  }










555













UNCOV

556



  void* pFileStore = getStateFileStore(pFileState);




×








UNCOV

557



  pCur = streamStateSessionSeekKeyCurrentPrev_rocksdb(pFileStore, pWinKey);




×








UNCOV

558



  if (!pCur) {





×








UNCOV

559



    return NULL;




×







560



  }











UNCOV

561



  pCur->buffIndex = -1;




×








UNCOV

562



  pCur->pStreamFileState = pFileState;




×








UNCOV

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













UNCOV

592



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




×








UNCOV

593



  if (!pCur) {





×







594



    return;




×







595



  }











UNCOV

596



  streamStateResetCur(pCur);




×








UNCOV

597



  pCur->buffIndex = 0;




×








UNCOV

598



  pCur->pStreamFileState = pFileState;




×







599



}










600













UNCOV

601



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




×







602



                                    SStreamStateCur** ppCur) {











UNCOV

603



  SSessionKey key = {.groupId = groupId};




×








UNCOV

604



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




×








UNCOV

605



  if (taosArrayGetSize(pWinStates) > 0 &&





×








UNCOV

606



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





×








UNCOV

607



    if (!(*ppCur)) {





×








UNCOV

608



      (*ppCur) = createStateCursor(pFileState);




×







609



    }











UNCOV

610



    transformCursor(pFileState, *ppCur);




×








UNCOV

611



  } else if (*ppCur) {





×








UNCOV

612



    (*ppCur)->buffIndex = -1;




×








UNCOV

613



    (*ppCur)->pStreamFileState = pFileState;




×







614



  }











UNCOV

615



}




×







616













UNCOV

617



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




×








UNCOV

618



  SArray*          pWinStates = NULL;




×








UNCOV

619



  int32_t          index = -1;




×








UNCOV

620



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




×








UNCOV

621



  if (pCur) {





×








UNCOV

622



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





×








UNCOV

623



      sessionWinStateMoveToNext(pCur);




×







624



    }











UNCOV

625



    return pCur;




×







626



  }










627













UNCOV

628



  void* pFileStore = getStateFileStore(pFileState);




×








UNCOV

629



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




×








UNCOV

630



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




×








UNCOV

631



  return pCur;




×







632



}










633













UNCOV

634



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




×








UNCOV

635



  SArray*          pWinStates = NULL;




×








UNCOV

636



  int32_t          index = -1;




×








UNCOV

637



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




×








UNCOV

638



  if (pCur) {





×








UNCOV

639



    sessionWinStateMoveToNext(pCur);




×








UNCOV

640



    return pCur;




×







641



  }










642













UNCOV

643



  void* pFileStore = getStateFileStore(pFileState);




×








UNCOV

644



  pCur = streamStateSessionSeekKeyNext_rocksdb(pFileStore, pWinKey);




×








UNCOV

645



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




×








UNCOV

646



  return pCur;




×







647



}










648













UNCOV

649



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




×








UNCOV

650



  SArray*          pWinStates = NULL;




×








UNCOV

651



  int32_t          index = -1;




×








UNCOV

652



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




×








UNCOV

653



  int32_t          resSize = getRowStateRowSize(pFileState);




×








UNCOV

654



  COUNT_TYPE       winCount = 0;




×








UNCOV

655



  if (pBuffCur) {





×








UNCOV

656



    while (index >= 0) {





×








UNCOV

657



      SRowBuffPos* pPos = taosArrayGetP(pWinStates, index);




×








UNCOV

658



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




×








UNCOV

659



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





×








UNCOV

660



        index--;




×








UNCOV

661



      } else if (index >= 0) {





×








UNCOV

662



        pBuffCur->buffIndex = index + 1;




×








UNCOV

663



        return pBuffCur;




×







664



      }










665



    }











UNCOV

666



    pBuffCur->buffIndex = 0;




×








UNCOV

667



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





×







668



    pBuffCur = createStateCursor(pFileState);




×







669



    if (pBuffCur == NULL) {





×







670



      return NULL;




×







671



    }










672



    pBuffCur->buffIndex = 0;




×







673



  }










674













UNCOV

675



  void*            pFileStore = getStateFileStore(pFileState);




×








UNCOV

676



  SStreamStateCur* pCur = streamStateSessionSeekKeyPrev_rocksdb(pFileStore, pWinKey);




×








UNCOV

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



  }











UNCOV

711



  return pBuffCur;




×







712



}










713













UNCOV

714



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




×








UNCOV

715



  if (!pCur) {





×








UNCOV

716



    return TSDB_CODE_FAILED;




×







717



  }











UNCOV

718



  int32_t code = TSDB_CODE_SUCCESS;




×







719













UNCOV

720



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




×








UNCOV

721



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




×








UNCOV

722



  SArray*    pWinStates = NULL;




×








UNCOV

723



  if (ppBuff) {





×








UNCOV

724



    pWinStates = (SArray*)(*ppBuff);




×







725



  }










726













UNCOV

727



  if (pCur->buffIndex >= 0) {





×








UNCOV

728



    int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

729



    if (pCur->buffIndex >= size) {





×








UNCOV

730



      return TSDB_CODE_FAILED;




×







731



    }











UNCOV

732



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




×








UNCOV

733



    if (pVal) {





×








UNCOV

734



      *pVal = pPos;




×







735



    }











UNCOV

736



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




×







737



  } else {











UNCOV

738



    void* pData = NULL;




×








UNCOV

739



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




×








UNCOV

740



    if (taosArrayGetSize(pWinStates) > 0 &&





×








UNCOV

741



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





×








UNCOV

742



      transformCursor(pCur->pStreamFileState, pCur);




×








UNCOV

743



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




×








UNCOV

744



      if (pVal) {





×







745



        *pVal = pPos;




×







746



      }











UNCOV

747



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




×








UNCOV

748



      code = TSDB_CODE_SUCCESS;




×








UNCOV

749



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





×








UNCOV

750



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




×








UNCOV

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



      }











UNCOV

757



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




×








UNCOV

758



      pNewPos->needFree = true;




×








UNCOV

759



      pNewPos->beFlushed = true;




×








UNCOV

760



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




×








UNCOV

761



      (*pVal) = pNewPos;




×







762



    }











UNCOV

763



    taosMemoryFreeClear(pData);





×







764



  }











UNCOV

765



  return code;




×







766



}










767













UNCOV

768



void sessionWinStateMoveToNext(SStreamStateCur* pCur) {




×








UNCOV

769



  qTrace("move cursor to next");





×








UNCOV

770



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





×








UNCOV

771



    pCur->buffIndex++;




×







772



  } else {











UNCOV

773



    streamStateCurNext_rocksdb(pCur);




×







774



  }











UNCOV

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













UNCOV

786



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




×







787



                                     range_cmpr_fn cmpFn) {











UNCOV

788



  SStreamStateCur* pCur = sessionWinStateSeekKeyCurrentPrev(pFileState, key);




×








UNCOV

789



  SSessionKey      tmpKey = *key;




×








UNCOV

790



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




×








UNCOV

791



  bool             hasCurrentPrev = true;




×








UNCOV

792



  if (code == TSDB_CODE_FAILED) {





×








UNCOV

793



    streamStateFreeCur(pCur);




×








UNCOV

794



    pCur = sessionWinStateSeekKeyNext(pFileState, key);




×








UNCOV

795



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




×








UNCOV

796



    hasCurrentPrev = false;




×







797



  }










798













UNCOV

799



  if (code == TSDB_CODE_FAILED) {





×








UNCOV

800



    code = TSDB_CODE_FAILED;




×








UNCOV

801



    goto _end;




×







802



  }










803













UNCOV

804



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





×








UNCOV

805



    *curKey = tmpKey;




×








UNCOV

806



    goto _end;




×








UNCOV

807



  } else if (!hasCurrentPrev) {





×








UNCOV

808



    code = TSDB_CODE_FAILED;




×








UNCOV

809



    goto _end;




×







810



  }










811













UNCOV

812



  sessionWinStateMoveToNext(pCur);




×








UNCOV

813



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




×








UNCOV

814



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





×








UNCOV

815



    *curKey = tmpKey;




×







816



  } else {











UNCOV

817



    code = TSDB_CODE_FAILED;




×







818



  }










819













UNCOV

820



_end:




×








UNCOV

821



  streamStateFreeCur(pCur);




×








UNCOV

822



  return code;




×







823



}










824













UNCOV

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) {











UNCOV

827



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

828



  int32_t lino = 0;




×








UNCOV

829



  (*pWinCode) = TSDB_CODE_SUCCESS;




×







830













UNCOV

831



  SSessionKey* pWinKey = key;




×








UNCOV

832



  TSKEY        gap = 0;




×








UNCOV

833



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

834



  SArray*      pWinStates = NULL;




×







835













UNCOV

836



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




×








UNCOV

837



  if (ppBuff) {





×








UNCOV

838



    pWinStates = (SArray*)(*ppBuff);




×







839



  } else {











UNCOV

840



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×








UNCOV

841



    if (!pWinStates) {





×







842



      code = terrno;




×







843



      QUERY_CHECK_CODE(code, lino, _end);





×







844



    }










845













UNCOV

846



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




×








UNCOV

847



    QUERY_CHECK_CODE(code, lino, _end);





×







848



  }










849













UNCOV

850



  TSKEY startTs = pWinKey->win.skey;




×








UNCOV

851



  TSKEY endTs = pWinKey->win.ekey;




×







852













UNCOV

853



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

854



  if (size == 0) {





×








UNCOV

855



    void*   pFileStore = getStateFileStore(pFileState);




×








UNCOV

856



    void*   p = NULL;




×








UNCOV

857



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




×








UNCOV

858



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





×








UNCOV

859



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




×








UNCOV

860



      if (!(*pVal)) {





×







861



        code = TSDB_CODE_OUT_OF_MEMORY;




×







862



        QUERY_CHECK_CODE(code, lino, _end);





×







863



      }










864













UNCOV

865



      (*pWinCode) = code_file;




×








UNCOV

866



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





×







867



             pWinKey->win.ekey, code_file);










868



    } else {











UNCOV

869



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




×








UNCOV

870



      (*pWinCode) = TSDB_CODE_FAILED;




×








UNCOV

871



      taosMemoryFree(p);





×








UNCOV

872



      QUERY_CHECK_CODE(code, lino, _end);





×







873



    }











UNCOV

874



    goto _end;




×







875



  }










876












877



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











UNCOV

878



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




×








UNCOV

879



  SRowBuffPos* pPos = NULL;




×








UNCOV

880



  int32_t      valSize = *pVLen;




×







881













UNCOV

882



  if (index >= 0) {





×








UNCOV

883



    pPos = taosArrayGetP(pWinStates, index);




×








UNCOV

884



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




×








UNCOV

885



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





×








UNCOV

886



      (*pVal) = pPos;




×








UNCOV

887



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




×








UNCOV

888



      pPos->beUsed = true;




×








UNCOV

889



      pPos->beFlushed = false;




×








UNCOV

890



      *key = *pDestWinKey;




×








UNCOV

891



      goto _end;




×







892



    }










893



  }










894













UNCOV

895



  if (index + 1 < size) {





×








UNCOV

896



    pPos = taosArrayGetP(pWinStates, index + 1);




×








UNCOV

897



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




×








UNCOV

898



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





×








UNCOV

899



        fn(pKeyData, stateKey) == true) {




×








UNCOV

900



      (*pVal) = pPos;




×








UNCOV

901



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




×








UNCOV

902



      pPos->beUsed = true;




×








UNCOV

903



      pPos->beFlushed = false;




×








UNCOV

904



      *key = *pDestWinKey;




×








UNCOV

905



      goto _end;




×







906



    }










907



  }










908













UNCOV

909



  if (index + 1 == 0) {





×








UNCOV

910



    if (!isDeteled(pFileState, endTs)) {





×








UNCOV

911



      void*   p = NULL;




×








UNCOV

912



      void*   pFileStore = getStateFileStore(pFileState);




×







913



      int32_t code_file =











UNCOV

914



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




×








UNCOV

915



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





×








UNCOV

916



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




×








UNCOV

917



        if (!(*pVal)) {





×







918



          code = TSDB_CODE_OUT_OF_MEMORY;




×








UNCOV

919



          QUERY_CHECK_CODE(code, lino, _end);





×







920



        }










921













UNCOV

922



        (*pWinCode) = code_file;




×








UNCOV

923



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





×







924



               pWinKey->win.ekey, code_file);











UNCOV

925



        goto _end;




×







926



      } else {











UNCOV

927



        taosMemoryFree(p);





×







928



      }










929



    }










930



  }










931













UNCOV

932



  if (index == size - 1) {





×








UNCOV

933



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




×








UNCOV

934



    QUERY_CHECK_CODE(code, lino, _end);





×







935













UNCOV

936



    (*pWinCode) = TSDB_CODE_FAILED;




×








UNCOV

937



    goto _end;




×







938



  }











UNCOV

939



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




×








UNCOV

940



  QUERY_CHECK_CODE(code, lino, _end);





×







941













UNCOV

942



  (*pWinCode) = TSDB_CODE_FAILED;




×







943













UNCOV

944



_end:




×








UNCOV

945



  if (code != TSDB_CODE_SUCCESS) {





×







946



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





×







947



  }











UNCOV

948



  return code;




×







949



}










950













UNCOV

951



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




×








UNCOV

952



  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext_rocksdb(pState, pKey);




×








UNCOV

953



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




×








UNCOV

954



  streamStateFreeCur(pCur);




×








UNCOV

955



  if (code == TSDB_CODE_SUCCESS) {





×







956



    return code;




×







957



  } else {











UNCOV

958



    pCur = streamStateSessionSeekKeyPrev_rocksdb(pState, pKey);




×







959



  }










960













UNCOV

961



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




×








UNCOV

962



  streamStateFreeCur(pCur);




×








UNCOV

963



  return code;




×







964



}










965













UNCOV

966



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




×







967



                              int32_t* pVLen, int32_t* pWinCount) {











UNCOV

968



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

969



  int32_t lino = 0;




×








UNCOV

970



  (*pWinCount) = TSDB_CODE_SUCCESS;




×







971













UNCOV

972



  SSessionKey* pWinKey = pKey;




×








UNCOV

973



  const TSKEY  gap = 0;




×








UNCOV

974



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

975



  SArray*      pWinStates = NULL;




×








UNCOV

976



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




×








UNCOV

977



  if (ppBuff) {





×








UNCOV

978



    pWinStates = (SArray*)(*ppBuff);




×







979



  } else {











UNCOV

980



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×








UNCOV

981



    if (!pWinStates) {





×







982



      code = terrno;




×







983



      QUERY_CHECK_CODE(code, lino, _end);





×







984



    }










985













UNCOV

986



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




×








UNCOV

987



    QUERY_CHECK_CODE(code, lino, _end);





×







988



  }










989













UNCOV

990



  TSKEY startTs = pWinKey->win.skey;




×








UNCOV

991



  TSKEY endTs = pWinKey->win.ekey;




×







992













UNCOV

993



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

994



  if (size == 0) {





×








UNCOV

995



    void* pFileStore = getStateFileStore(pFileState);




×








UNCOV

996



    void* pRockVal = NULL;




×








UNCOV

997



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




×








UNCOV

998



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





×








UNCOV

999



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





×







1000



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











UNCOV

1001



      if ((*pWinCount) == TSDB_CODE_SUCCESS) {





×








UNCOV

1002



        int32_t     valSize = *pVLen;




×








UNCOV

1003



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




×








UNCOV

1004



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





×








UNCOV

1005



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




×








UNCOV

1006



          if (!(*pVal)) {





×







1007



            code = TSDB_CODE_OUT_OF_MEMORY;




×







1008



            QUERY_CHECK_CODE(code, lino, _end);





×







1009



          }










1010













UNCOV

1011



          goto _end;




×







1012



        }










1013



      }











UNCOV

1014



      pWinKey->win.skey = startTs;




×








UNCOV

1015



      pWinKey->win.ekey = endTs;




×








UNCOV

1016



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




×








UNCOV

1017



      taosMemoryFree(pRockVal);





×








UNCOV

1018



      if (!(*pVal)) {





×







1019



        code = TSDB_CODE_OUT_OF_MEMORY;




×







1020



        QUERY_CHECK_CODE(code, lino, _end);





×







1021



      }










1022



    } else {











UNCOV

1023



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




×








UNCOV

1024



      QUERY_CHECK_CODE(code, lino, _end);





×







1025













UNCOV

1026



      (*pWinCount) = TSDB_CODE_FAILED;




×







1027



    }











UNCOV

1028



    goto _end;




×







1029



  }










1030












1031



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











UNCOV

1032



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




×








UNCOV

1033



  SRowBuffPos* pPos = NULL;




×








UNCOV

1034



  int32_t      valSize = *pVLen;




×







1035













UNCOV

1036



  if (index >= 0) {





×








UNCOV

1037



    pPos = taosArrayGetP(pWinStates, index);




×








UNCOV

1038



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




×








UNCOV

1039



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





×








UNCOV

1040



      (*pVal) = pPos;




×








UNCOV

1041



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




×








UNCOV

1042



      pPos->beUsed = true;




×








UNCOV

1043



      pPos->beFlushed = false;




×








UNCOV

1044



      *pWinKey = *pDestWinKey;




×








UNCOV

1045



      goto _end;




×







1046



    }










1047



  }










1048













UNCOV

1049



  if (index == -1) {





×








UNCOV

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













UNCOV

1076



  if (index + 1 < size) {





×








UNCOV

1077



    pPos = taosArrayGetP(pWinStates, index + 1);




×








UNCOV

1078



    (*pVal) = pPos;




×








UNCOV

1079



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




×








UNCOV

1080



    pPos->beUsed = true;




×








UNCOV

1081



    pPos->beFlushed = false;




×








UNCOV

1082



    *pWinKey = *pDestWinKey;




×








UNCOV

1083



    goto _end;




×







1084



  }










1085













UNCOV

1086



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




×








UNCOV

1087



  QUERY_CHECK_CODE(code, lino, _end);





×







1088













UNCOV

1089



  (*pWinCount) = TSDB_CODE_FAILED;




×







1090













UNCOV

1091



_end:




×








UNCOV

1092



  if (code != TSDB_CODE_SUCCESS) {





×







1093



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





×







1094



  }











UNCOV

1095



  return code;




×







1096



}










1097













UNCOV

1098



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




×







1099



                                 int32_t* pVLen) {











UNCOV

1100



  SSessionKey* pWinKey = pKey;




×








UNCOV

1101



  const TSKEY  gap = 0;




×








UNCOV

1102



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

1103



  int32_t      lino = 0;




×








UNCOV

1104



  SSHashObj*   pSessionBuff = getRowStateBuff(pFileState);




×








UNCOV

1105



  SArray*      pWinStates = NULL;




×








UNCOV

1106



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




×








UNCOV

1107



  if (ppBuff) {





×








UNCOV

1108



    pWinStates = (SArray*)(*ppBuff);




×







1109



  } else {











UNCOV

1110



    pWinStates = taosArrayInit(16, POINTER_BYTES);




×








UNCOV

1111



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




×








UNCOV

1112



    QUERY_CHECK_CODE(code, lino, _end);





×







1113



  }










1114













UNCOV

1115



  TSKEY startTs = pWinKey->win.skey;




×








UNCOV

1116



  TSKEY endTs = pWinKey->win.ekey;




×







1117













UNCOV

1118



  int32_t size = taosArrayGetSize(pWinStates);




×








UNCOV

1119



  if (size == 0) {





×








UNCOV

1120



    void* pFileStore = getStateFileStore(pFileState);




×








UNCOV

1121



    void* pRockVal = NULL;




×







1122













UNCOV

1123



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




×








UNCOV

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 {











UNCOV

1140



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




×








UNCOV

1141



      taosMemoryFree(pRockVal);





×








UNCOV

1142



      QUERY_CHECK_CODE(code, lino, _end);





×







1143



    }










1144



  } else {











UNCOV

1145



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




×








UNCOV

1146



    QUERY_CHECK_CODE(code, lino, _end);





×







1147



  }










1148













UNCOV

1149



_end:




×








UNCOV

1150



  if (code != TSDB_CODE_SUCCESS) {





×







1151



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





×







1152



  }











UNCOV

1153



  return code;




×







1154



}
























    

  • 

    •