#3533

Committed 20 Nov 2024 07:11AM UTC coverage: 58.848% (-1.9%) from 60.78%

Build # #3533

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28823 from taosdata/fix/3.0/TD-32587

fix:[TD-32587]fix stmt segmentation fault

Run Details

115578 of 252434 branches covered (45.79%)

Branch coverage included in aggregate %.

1 of 4 new or added lines in 1 file covered. (25.0%)

8038 existing lines in 233 files now uncovered.

194926 of 275199 relevant lines covered (70.83%)

1494459.59 hits per line

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

74.06

/source/libs/stream/src/streamState.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "streamState.h"
#include "executor.h"
#include "osMemory.h"
#include "rocksdb/c.h"
#include "streamBackendRocksdb.h"
#include "streamInt.h"
#include "tcoding.h"
#include "tcommon.h"
#include "tcompare.h"
#include "tref.h"

#define MAX_TABLE_NAME_NUM 200000

int sessionRangeKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pWin2) {
  if (pWin1->groupId > pWin2->groupId) {
    return 1;
  } else if (pWin1->groupId < pWin2->groupId) {
    return -1;
  }

  if (pWin1->win.skey > pWin2->win.ekey) {
    return 1;
  } else if (pWin1->win.ekey < pWin2->win.skey) {
    return -1;
  }

  return 0;
}

int countRangeKeyEqual(const SSessionKey* pWin1, const SSessionKey* pWin2) {
  if (pWin1->groupId == pWin2->groupId && pWin1->win.skey <= pWin2->win.skey && pWin2->win.skey <= pWin1->win.ekey) {
    return 0;
  }

  return 1;
}

int sessionWinKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pWin2) {
  if (pWin1->groupId > pWin2->groupId) {
    return 1;
  } else if (pWin1->groupId < pWin2->groupId) {
    return -1;
  }

  if (pWin1->win.skey > pWin2->win.skey) {
    return 1;
  } else if (pWin1->win.skey < pWin2->win.skey) {
    return -1;
  }

  if (pWin1->win.ekey > pWin2->win.ekey) {
    return 1;
  } else if (pWin1->win.ekey < pWin2->win.ekey) {
    return -1;
  }

  return 0;
}

int stateSessionKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
  SStateSessionKey* pWin1 = (SStateSessionKey*)pKey1;
  SStateSessionKey* pWin2 = (SStateSessionKey*)pKey2;

  if (pWin1->opNum > pWin2->opNum) {
    return 1;
  } else if (pWin1->opNum < pWin2->opNum) {
    return -1;
  }

  return sessionWinKeyCmpr(&pWin1->key, &pWin2->key);
}

int stateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
  SStateKey* pWin1 = (SStateKey*)pKey1;
  SStateKey* pWin2 = (SStateKey*)pKey2;

  if (pWin1->opNum > pWin2->opNum) {
    return 1;
  } else if (pWin1->opNum < pWin2->opNum) {
    return -1;
  }

  return winKeyCmprImpl(&pWin1->key, &pWin2->key);
}

SStreamState* streamStateOpen(const char* path, void* pTask, int64_t streamId, int32_t taskId) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState));
  stDebug("open stream state %p, %s", pState, path);

  if (pState == NULL) {
    code = terrno;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  pState->pTdbState = taosMemoryCalloc(1, sizeof(STdbState));
  if (pState->pTdbState == NULL) {
    streamStateDestroy(pState, true);
    code = terrno;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  SStreamTask* pStreamTask = pTask;
  pState->streamId = streamId;
  pState->taskId = taskId;
  TAOS_UNUSED(tsnprintf(pState->pTdbState->idstr, sizeof(pState->pTdbState->idstr), "0x%" PRIx64 "-0x%x", pState->streamId, pState->taskId));

  code = streamTaskSetDb(pStreamTask->pMeta, pTask, pState->pTdbState->idstr);
  QUERY_CHECK_CODE(code, lino, _end);

  SStreamMeta* pMeta = pStreamTask->pMeta;
  pState->pTdbState->pOwner = pTask;
  pState->pFileState = NULL;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT);
  pState->parNameMap = tSimpleHashInit(1024, hashFn);
  QUERY_CHECK_NULL(pState->parNameMap, code, lino, _end, terrno);

  stInfo("open state %p on backend %p 0x%" PRIx64 "-%d succ", pState, pMeta->streamBackend, pState->streamId,
         pState->taskId);
  return pState;

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("0x%x %s failed at line %d since %s", taskId, __func__, lino, tstrerror(code));
  }
  return NULL;
}

int32_t streamStateDelTaskDb(SStreamState* pState) {
  SStreamTask* pTask = pState->pTdbState->pOwner;
  taskDbRemoveRef(pTask->pBackend);
  taosMemoryFree(pTask);
  return 0;
}
void streamStateClose(SStreamState* pState, bool remove) {
  SStreamTask* pTask = pState->pTdbState->pOwner;
  streamStateDestroy(pState, remove);
}

int32_t streamStateBegin(SStreamState* pState) { return 0; }

void streamStateCommit(SStreamState* pState) {
  if (pState->pFileState) {
    SStreamSnapshot* pShot = getSnapshot(pState->pFileState);
    flushSnapshot(pState->pFileState, pShot, true);
  }
}

int32_t streamStateFuncPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  void*   pVal = NULL;
  int32_t len = getRowStateRowSize(pState->pFileState);
  int32_t tmpLen = len;
  code = getFunctionRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), &pVal, &tmpLen);
  QUERY_CHECK_CODE(code, lino, _end);

  char*   buf = ((SRowBuffPos*)pVal)->pRowBuff;
  int32_t rowSize = streamFileStateGetSelectRowSize(pState->pFileState);
  memcpy(buf + len - rowSize, value, vLen);

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}
int32_t streamStateFuncGet(SStreamState* pState, const SWinKey* key, void** ppVal, int32_t* pVLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  void*   pVal = NULL;
  int32_t len = getRowStateRowSize(pState->pFileState);
  int32_t tmpLen = len;
  code = getFunctionRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), (void**)(&pVal), &tmpLen);
  QUERY_CHECK_CODE(code, lino, _end);

  char*   buf = ((SRowBuffPos*)pVal)->pRowBuff;
  int32_t rowSize = streamFileStateGetSelectRowSize(pState->pFileState);
  *ppVal = buf + len - rowSize;
  streamStateReleaseBuf(pState, pVal, false);

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t streamStatePut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) { return 0; }

int32_t streamStateGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen, int32_t* pWinCode) {
  return addRowBuffIfNotExist(pState->pFileState, (void*)key, sizeof(SWinKey), pVal, pVLen, pWinCode);
}

bool streamStateCheck(SStreamState* pState, const SWinKey* key) {
  return hasRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey));
}

int32_t streamStateGetByPos(SStreamState* pState, void* pos, void** pVal) {
  int32_t code = getRowBuffByPos(pState->pFileState, pos, pVal);
  streamStateReleaseBuf(pState, pos, false);
  return code;
}

void streamStateDel(SStreamState* pState, const SWinKey* key) {
  deleteRowBuff(pState->pFileState, key, sizeof(SWinKey));
}

int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen) {
  return streamStateFillPut_rocksdb(pState, key, value, vLen);
}

int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen, int32_t* pWinCode) {
  if (pState->pFileState) {
    return getRowBuff(pState->pFileState, (void*)key, sizeof(SWinKey), pVal, pVLen, pWinCode);
  }
  return streamStateFillGet_rocksdb(pState, key, pVal, pVLen);
}

int32_t streamStateFillAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen,
                                     int32_t* pWinCode) {
  return getHashSortRowBuff(pState->pFileState, key, pVal, pVLen, pWinCode);
}

int32_t streamStateFillGetNext(SStreamState* pState, const SWinKey* pKey, SWinKey* pResKey, void** pVal, int32_t* pVLen,
                               int32_t* pWinCode) {
  return getHashSortNextRow(pState->pFileState, pKey, pResKey, pVal, pVLen, pWinCode);
}

int32_t streamStateFillGetPrev(SStreamState* pState, const SWinKey* pKey, SWinKey* pResKey, void** pVal, int32_t* pVLen,
                               int32_t* pWinCode) {
  return getHashSortPrevRow(pState->pFileState, pKey, pResKey, pVal, pVLen, pWinCode);
}

void streamStateFillDel(SStreamState* pState, const SWinKey* key) {
  int32_t code = streamStateFillDel_rocksdb(pState, key);
  qTrace("%s at line %d res %d", __func__, __LINE__, code);
}

void streamStateClear(SStreamState* pState) { streamFileStateClear(pState->pFileState); }

void streamStateSetNumber(SStreamState* pState, int32_t number, int32_t tsIdex) {
  pState->number = number;
  pState->tsIndex = tsIdex;
}

void streamStateSaveInfo(SStreamState* pState, void* pKey, int32_t keyLen, void* pVal, int32_t vLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  char* cfName = "default";
  void* batch = streamStateCreateBatch();
  code = streamStatePutBatch(pState, cfName, batch, pKey, pVal, vLen, 0);
  QUERY_CHECK_CODE(code, lino, _end);

  code = streamStatePutBatch_rocksdb(pState, batch);
  QUERY_CHECK_CODE(code, lino, _end);

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  streamStateDestroyBatch(batch);
}

int32_t streamStateGetInfo(SStreamState* pState, void* pKey, int32_t keyLen, void** pVal, int32_t* pLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  code = streamDefaultGet_rocksdb(pState, pKey, pVal, pLen);
  return code;
}

int32_t streamStateAddIfNotExist(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen,
                                 int32_t* pWinCode) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  code = streamStateGet(pState, key, pVal, pVLen, pWinCode);
  QUERY_CHECK_CODE(code, lino, _end);

  SSHashObj* pSearchBuff = getSearchBuff(pState->pFileState);
  if (pSearchBuff != NULL) {
    SArray* pWinStates = NULL;
    code = addArrayBuffIfNotExist(pSearchBuff, key->groupId, &pWinStates);
    QUERY_CHECK_CODE(code, lino, _end);
    code = addSearchItem(pState->pFileState, pWinStates, key);
    QUERY_CHECK_CODE(code, lino, _end);
  }
_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

void streamStateReleaseBuf(SStreamState* pState, void* pVal, bool used) {
  if (!pVal) {
    return;
  }
  streamFileStateReleaseBuff(pState->pFileState, pVal, used);
}

void streamStateClearBuff(SStreamState* pState, void* pVal) { streamFileStateClearBuff(pState->pFileState, pVal); }

SStreamStateCur* streamStateFillGetCur(SStreamState* pState, const SWinKey* key) {
  return streamStateFillGetCur_rocksdb(pState, key);
}

SStreamStateCur* streamStateGetAndCheckCur(SStreamState* pState, SWinKey* key) {
  return streamStateGetAndCheckCur_rocksdb(pState, key);
}

int32_t streamStateGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
  return streamStateGetKVByCur_rocksdb(getStateFileStore(pCur->pStreamFileState), pCur, pKey, pVal, pVLen);
}

int32_t streamStateFillGetKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
  return streamStateFillGetKVByCur_rocksdb(pCur, pKey, pVal, pVLen);
}

int32_t streamStateFillGetGroupKVByCur(SStreamStateCur* pCur, SWinKey* pKey, const void** pVal, int32_t* pVLen) {
  return streamStateFillGetGroupKVByCur_rocksdb(pCur, pKey, pVal, pVLen);
}

SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key) {
  return streamStateSeekKeyNext_rocksdb(pState, key);
}

SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey* key) {
  return streamStateFillSeekKeyNext_rocksdb(pState, key);
}

SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey* key) {
  return streamStateFillSeekKeyPrev_rocksdb(pState, key);
}

void streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur) { sessionWinStateMoveToNext(pCur); }

void streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur) {
  qTrace("move cursor to next");
  streamStateCurPrev_rocksdb(pCur);
}

void streamStateResetCur(SStreamStateCur* pCur) {
  if (!pCur) {
    return;
  }
  if (pCur->iter) rocksdb_iter_destroy(pCur->iter);
  if (pCur->snapshot) rocksdb_release_snapshot(pCur->db, pCur->snapshot);
  if (pCur->readOpt) rocksdb_readoptions_destroy(pCur->readOpt);

  memset(pCur, 0, sizeof(SStreamStateCur));

  pCur->buffIndex = -1;
}

void streamStateFreeCur(SStreamStateCur* pCur) {
  if (!pCur) {
    return;
  }
  streamStateResetCur(pCur);
  taosMemoryFree(pCur);
}

void streamStateFreeVal(void* val) { taosMemoryFree(val); }

int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, void* value, int32_t vLen) {
  int32_t      code = TSDB_CODE_SUCCESS;
  int32_t      lino = 0;
  SRowBuffPos* pos = (SRowBuffPos*)value;
  if (pos->needFree) {
    if (isFlushedState(pState->pFileState, key->win.ekey, 0)) {
      if (!pos->pRowBuff) {
        goto _end;
      }
      code = streamStateSessionPut_rocksdb(pState, key, pos->pRowBuff, vLen);
      QUERY_CHECK_CODE(code, lino, _end);

      streamStateReleaseBuf(pState, pos, true);
      code = putFreeBuff(pState->pFileState, pos);
      QUERY_CHECK_CODE(code, lino, _end);

      stDebug("===stream===save skey:%" PRId64 ", ekey:%" PRId64 ", groupId:%" PRIu64 ".code:%d", key->win.skey,
              key->win.ekey, key->groupId, code);
    } else {
      pos->beFlushed = false;
      code = putSessionWinResultBuff(pState->pFileState, value);
      QUERY_CHECK_CODE(code, lino, _end);
    }
  }

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t streamStateSessionAllocWinBuffByNextPosition(SStreamState* pState, SStreamStateCur* pCur,
                                                     const SSessionKey* pKey, void** pVal, int32_t* pVLen) {
  return allocSessioncWinBuffByNextPosition(pState->pFileState, pCur, pKey, pVal, pVLen);
}

int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen, int32_t* pWinCode) {
  return getSessionFlushedBuff(pState->pFileState, key, pVal, pVLen, pWinCode);
}

void streamStateSessionDel(SStreamState* pState, const SSessionKey* key) {
  qDebug("===stream===delete skey:%" PRId64 ", ekey:%" PRId64 ", groupId:%" PRIu64, key->win.skey, key->win.ekey,
         key->groupId);
  deleteRowBuff(pState->pFileState, key, sizeof(SSessionKey));
}

void streamStateSessionReset(SStreamState* pState, void* pVal) {
  int32_t len = getRowStateRowSize(pState->pFileState);
  memset(pVal, 0, len);
}

SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, const SSessionKey* key) {
  return sessionWinStateSeekKeyCurrentPrev(pState->pFileState, key);
}

SStreamStateCur* streamStateSessionSeekKeyCurrentNext(SStreamState* pState, const SSessionKey* key) {
  return sessionWinStateSeekKeyCurrentNext(pState->pFileState, key);
}

SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSessionKey* key) {
  return sessionWinStateSeekKeyNext(pState->pFileState, key);
}

SStreamStateCur* streamStateCountSeekKeyPrev(SStreamState* pState, const SSessionKey* key, COUNT_TYPE count) {
  return countWinStateSeekKeyPrev(pState->pFileState, key, count);
}

int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen) {
  return sessionWinStateGetKVByCur(pCur, pKey, pVal, pVLen);
}

void streamStateSessionClear(SStreamState* pState) {
  sessionWinStateClear(pState->pFileState);
  streamStateSessionClear_rocksdb(pState);
}

int32_t streamStateSessionGetKeyByRange(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
  return sessionWinStateGetKeyByRange(pState->pFileState, key, curKey, sessionRangeKeyCmpr);
}

int32_t streamStateCountGetKeyByRange(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
  return sessionWinStateGetKeyByRange(pState->pFileState, key, curKey, countRangeKeyEqual);
}

int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal, int32_t* pVLen,
                                        int32_t* pWinCode) {
  return getSessionWinResultBuff(pState->pFileState, key, gap, pVal, pVLen, pWinCode);
}

int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
                                      state_key_cmpr_fn fn, void** pVal, int32_t* pVLen, int32_t* pWinCode) {
  return getStateWinResultBuff(pState->pFileState, key, pKeyData, keyDataLen, fn, pVal, pVLen, pWinCode);
}

int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char tbname[TSDB_TABLE_NAME_LEN]) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (tSimpleHashGet(pState->parNameMap, &groupId, sizeof(int64_t)) == NULL) {
    if (tSimpleHashGetSize(pState->parNameMap) < MAX_TABLE_NAME_NUM) {
      code = tSimpleHashPut(pState->parNameMap, &groupId, sizeof(int64_t), tbname, TSDB_TABLE_NAME_LEN);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    code = streamStatePutParName_rocksdb(pState, groupId, tbname);
    QUERY_CHECK_CODE(code, lino, _end);
  }

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal, bool onlyCache, int32_t* pWinCode) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  void*   pStr = tSimpleHashGet(pState->parNameMap, &groupId, sizeof(int64_t));
  if (!pStr) {
    if (onlyCache && tSimpleHashGetSize(pState->parNameMap) < MAX_TABLE_NAME_NUM) {
      (*pWinCode) = TSDB_CODE_FAILED;
      goto _end;
    }
    (*pWinCode) = streamStateGetParName_rocksdb(pState, groupId, pVal);
    if ((*pWinCode) == TSDB_CODE_SUCCESS && tSimpleHashGetSize(pState->parNameMap) < MAX_TABLE_NAME_NUM) {
      code = tSimpleHashPut(pState->parNameMap, &groupId, sizeof(int64_t), *pVal, TSDB_TABLE_NAME_LEN);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    goto _end;
  }
  *pVal = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
  if (!(*pVal)) {
    code = terrno;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  memcpy(*pVal, pStr, TSDB_TABLE_NAME_LEN);
  (*pWinCode) = TSDB_CODE_SUCCESS;

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t streamStateDeleteParName(SStreamState* pState, int64_t groupId) {
  int32_t code = tSimpleHashRemove(pState->parNameMap, &groupId, sizeof(int64_t));
  if (TSDB_CODE_SUCCESS != code) {
    qWarn("failed to remove parname from cache, code:%d", code);
  }
  code = streamStateDeleteParName_rocksdb(pState, groupId);
  if (TSDB_CODE_SUCCESS != code) {
    qWarn("failed to remove parname from rocksdb, code:%d", code);
  }
  return TSDB_CODE_SUCCESS;
}

void streamStateDestroy(SStreamState* pState, bool remove) {
  streamFileStateDestroy(pState->pFileState);
  // streamStateDestroy_rocksdb(pState, remove);
  tSimpleHashCleanup(pState->parNameMap);
  // do nothong
  taosMemoryFreeClear(pState->pTdbState);
  taosMemoryFreeClear(pState);
}

int32_t streamStateDeleteCheckPoint(SStreamState* pState, TSKEY mark) {
  return deleteExpiredCheckPoint(pState->pFileState, mark);
}

void streamStateReloadInfo(SStreamState* pState, TSKEY ts) { streamFileStateReloadInfo(pState->pFileState, ts); }

void streamStateCopyBackend(SStreamState* src, SStreamState* dst) {
  dst->pFileState = src->pFileState;
  dst->parNameMap = src->parNameMap;
  dst->number = src->number;
  dst->taskId = src->taskId;
  dst->streamId = src->streamId;
  if (dst->pTdbState == NULL) {
    dst->pTdbState = taosMemoryCalloc(1, sizeof(STdbState));
    dst->pTdbState->pOwner = taosMemoryCalloc(1, sizeof(SStreamTask));
  }
  dst->dump = 1;
  dst->pTdbState->pOwner->pBackend = src->pTdbState->pOwner->pBackend;
  dst->pResultRowStore.resultRowPut = src->pResultRowStore.resultRowPut;
  dst->pResultRowStore.resultRowGet = src->pResultRowStore.resultRowGet;
  dst->pExprSupp = src->pExprSupp;
  return;
}
SStreamStateCur* createStreamStateCursor() {
  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
  if (pCur == NULL) {
    return NULL;
  }
  pCur->buffIndex = -1;
  return pCur;
}

// count window
int32_t streamStateCountWinAddIfNotExist(SStreamState* pState, SSessionKey* pKey, COUNT_TYPE winCount, void** ppVal,
                                         int32_t* pVLen, int32_t* pWinCode) {
  return getCountWinResultBuff(pState->pFileState, pKey, winCount, ppVal, pVLen, pWinCode);
}

int32_t streamStateCountWinAdd(SStreamState* pState, SSessionKey* pKey, COUNT_TYPE winCount, void** pVal, int32_t* pVLen) {
  return createCountWinResultBuff(pState->pFileState, pKey, winCount, pVal, pVLen);
}

int32_t streamStateGroupPut(SStreamState* pState, int64_t groupId, void* value, int32_t vLen) {
  return streamFileStateGroupPut(pState->pFileState, groupId, value, vLen);
}

SStreamStateCur* streamStateGroupGetCur(SStreamState* pState) {
  SStreamStateCur* pCur = createStateCursor(pState->pFileState);
  pCur->hashIter = 0;
  pCur->pHashData = NULL;
  SSHashObj* pMap = getGroupIdCache(pState->pFileState);
  pCur->pHashData = tSimpleHashIterate(pMap, pCur->pHashData, &pCur->hashIter);
  if (pCur->pHashData == NULL) {
    pCur->hashIter = -1;
    streamStateParTagSeekKeyNext_rocksdb(pState, INT64_MIN, pCur);
  }
  return pCur;
}

void streamStateGroupCurNext(SStreamStateCur* pCur) {
  streamFileStateGroupCurNext(pCur);
}

int32_t streamStateGroupGetKVByCur(SStreamStateCur* pCur, int64_t* pKey, void** pVal, int32_t* pVLen) {
  if (pVal != NULL) {
    return -1;
  }
  return streamFileStateGroupGetKVByCur(pCur, pKey, pVal, pVLen);
}

void streamStateClearExpiredState(SStreamState* pState) {
  clearExpiredState(pState->pFileState);
}

void streamStateSetFillInfo(SStreamState* pState) {
  setFillInfo(pState->pFileState);
}

int32_t streamStateGetPrev(SStreamState* pState, const SWinKey* pKey, SWinKey* pResKey, void** pVal, int32_t* pVLen,
                           int32_t* pWinCode) {
  return getRowStatePrevRow(pState->pFileState, pKey, pResKey, pVal, pVLen, pWinCode);
}

taosdata / TDengine / #3533

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

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