#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

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

60.62

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

#include "query.h"
#include "streamBackendRocksdb.h"
#include "taos.h"
#include "tcommon.h"
#include "tcompare.h"
#include "thash.h"
#include "tsimplehash.h"

#define FLUSH_RATIO                    0.5
#define FLUSH_NUM                      4
#define DEFAULT_MAX_STREAM_BUFFER_SIZE (128 * 1024 * 1024)
#define MIN_NUM_OF_ROW_BUFF            10240
#define MIN_NUM_OF_RECOVER_ROW_BUFF    128
#define MIN_NUM_SEARCH_BUCKET          128
#define MAX_ARRAY_SIZE                 1024
#define MAX_GROUP_ID_NUM               200000
#define NUM_OF_CACHE_WIN               64
#define MAX_NUM_OF_CACHE_WIN           128
#define MIN_NUM_OF_SORT_CACHE_WIN      40960
#define BATCH_LIMIT                    256

#define DEFAULT_STATE_MAP_CAPACITY 10240
#define MAX_STATE_MAP_SIZE         10240000

#define SET_TSDATA_FLAG(ptr, len)   ((*(char*)POINTER_SHIFT(ptr, (len - 1))) |= 1)
#define UNSET_TSDATA_FLAG(ptr, len) ((*(char*)POINTER_SHIFT(ptr, (len - 1))) &= 0)
#define HAS_TSDATA_FLAG(ptr, len)   ((*(char*)POINTER_SHIFT(ptr, (len - 1))) & 1)

#define TASK_KEY               "streamFileState"
#define STREAM_STATE_INFO_NAME "StreamStateCheckPoint"

struct SStreamFileState {
  SList*     usedBuffs;
  SList*     freeBuffs;
  void*      rowStateBuff;
  void*      pFileStore;
  int32_t    rowSize;
  int32_t    selectivityRowSize;
  int32_t    keyLen;
  uint64_t   preCheckPointVersion;
  uint64_t   checkPointVersion;
  TSKEY      maxTs;
  TSKEY      deleteMark;
  TSKEY      flushMark;
  uint64_t   maxRowCount;
  uint64_t   curRowCount;
  GetTsFun   getTs;
  char*      id;
  char*      cfName;
  void*      searchBuff;
  SSHashObj* pGroupIdMap;
  bool       hasFillCatch;
  SSHashObj* pRecFlagMap;

  _state_buff_cleanup_fn         stateBuffCleanupFn;
  _state_buff_remove_fn          stateBuffRemoveFn;
  _state_buff_remove_by_pos_fn   stateBuffRemoveByPosFn;
  _state_buff_create_statekey_fn stateBuffCreateStateKeyFn;

  _state_file_remove_fn stateFileRemoveFn;
  _state_file_get_fn    stateFileGetFn;

  _state_fun_get_fn stateFunctionGetFn;
};

typedef SRowBuffPos SRowBuffInfo;

int fillStateKeyCompare(const void* pWin1, const void* pDatas, int pos) {
  SWinKey* pWin2 = taosArrayGet(pDatas, pos);
  return winKeyCmprImpl((SWinKey*)pWin1, pWin2);
}

int fillTSKeyCompare(const void* pKey1, const void* pDatas, int pos) {
  SWinKey* pWin1 = (SWinKey*)pKey1;
  SWinKey* pWin2 = taosArrayGet(pDatas, pos);
  if (pWin1->ts > pWin2->ts) {
    return 1;
  } else if (pWin1->ts < pWin2->ts) {
    return -1;
  }

  return 0;
}

int32_t stateHashBuffRemoveFn(void* pBuff, const void* pKey, size_t keyLen) {
  SRowBuffPos** pos = tSimpleHashGet(pBuff, pKey, keyLen);
  if (pos) {
    (*pos)->beFlushed = true;
    (*pos)->invalid = true;
  }
  return tSimpleHashRemove(pBuff, pKey, keyLen);
}

void stateHashBuffRemoveByPosFn(SStreamFileState* pFileState, SRowBuffPos* pPos) {
  size_t        keyLen = pFileState->keyLen;
  SRowBuffPos** ppPos = tSimpleHashGet(pFileState->rowStateBuff, pPos->pKey, keyLen);
  if (ppPos) {
    if ((*ppPos) == pPos) {
      int32_t tmpRes = tSimpleHashRemove(pFileState->rowStateBuff, pPos->pKey, keyLen);
      qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
    }
  }
}

void stateHashBuffClearFn(void* pBuff) { tSimpleHashClear(pBuff); }

void stateHashBuffCleanupFn(void* pBuff) { tSimpleHashCleanup(pBuff); }

int32_t intervalFileRemoveFn(SStreamFileState* pFileState, const void* pKey) {
  return streamStateDel_rocksdb(pFileState->pFileStore, pKey);
}

int32_t intervalFileGetFn(SStreamFileState* pFileState, void* pKey, void** data, int32_t* pDataLen) {
  return streamStateGet_rocksdb(pFileState->pFileStore, pKey, data, pDataLen);
}

void* intervalCreateStateKey(SRowBuffPos* pPos, int64_t num) {
  SStateKey* pStateKey = taosMemoryCalloc(1, sizeof(SStateKey));
  if (pStateKey == NULL) {
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
    return NULL;
  }
  SWinKey* pWinKey = pPos->pKey;
  pStateKey->key = *pWinKey;
  pStateKey->opNum = num;
  return pStateKey;
}

void* defaultCreateStateKey(SRowBuffPos* pPos, int64_t num) {
  SWinKey* pStateKey = taosMemoryCalloc(1, sizeof(SWinKey));
  if (pStateKey == NULL) {
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
    return NULL;
  }
  SWinKey* pWinKey = pPos->pKey;
  *pStateKey = *pWinKey;
  return pStateKey;
}

int32_t sessionFileRemoveFn(SStreamFileState* pFileState, const void* pKey) {
  return streamStateSessionDel_rocksdb(pFileState->pFileStore, pKey);
}

int32_t sessionFileGetFn(SStreamFileState* pFileState, void* pKey, void** data, int32_t* pDataLen) {
  return streamStateSessionGet_rocksdb(pFileState->pFileStore, pKey, data, pDataLen);
}

void* sessionCreateStateKey(SRowBuffPos* pPos, int64_t num) {
  SStateSessionKey* pStateKey = taosMemoryCalloc(1, sizeof(SStateSessionKey));
  if (pStateKey == NULL) {
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
    return NULL;
  }
  SSessionKey* pWinKey = pPos->pKey;
  pStateKey->key = *pWinKey;
  pStateKey->opNum = num;
  return pStateKey;
}

static void streamFileStateDecode(TSKEY* pKey, void* pBuff, int32_t len) { pBuff = taosDecodeFixedI64(pBuff, pKey); }

static int32_t streamFileStateEncode(TSKEY* pKey, void** pVal, int32_t* pLen) {
  *pLen = sizeof(TSKEY);
  (*pVal) = taosMemoryCalloc(1, *pLen);
  if ((*pVal) == NULL) {
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
    return terrno;
  }
  void*   buff = *pVal;
  int32_t tmp = taosEncodeFixedI64(&buff, *pKey);
  return TSDB_CODE_SUCCESS;
}

int32_t streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize, GetTsFun fp,
                            void* pFile, TSKEY delMark, const char* taskId, int64_t checkpointId, int8_t type,
                            SStreamFileState** ppFileState) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (memSize <= 0) {
    memSize = DEFAULT_MAX_STREAM_BUFFER_SIZE;
  }
  if (rowSize == 0) {
    code = TSDB_CODE_INVALID_PARA;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  SStreamFileState* pFileState = taosMemoryCalloc(1, sizeof(SStreamFileState));
  QUERY_CHECK_NULL(pFileState, code, lino, _end, terrno);

  rowSize += selectRowSize;
  pFileState->maxRowCount = TMAX((uint64_t)memSize / rowSize, FLUSH_NUM * 2);
  pFileState->usedBuffs = tdListNew(POINTER_BYTES);
  QUERY_CHECK_NULL(pFileState->usedBuffs, code, lino, _end, terrno);

  pFileState->freeBuffs = tdListNew(POINTER_BYTES);
  QUERY_CHECK_NULL(pFileState->freeBuffs, code, lino, _end, terrno);

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  int32_t    cap = TMIN(MIN_NUM_OF_ROW_BUFF, pFileState->maxRowCount);
  if (type == STREAM_STATE_BUFF_HASH || type == STREAM_STATE_BUFF_HASH_SEARCH) {
    pFileState->rowStateBuff = tSimpleHashInit(cap, hashFn);
    pFileState->stateBuffCleanupFn = stateHashBuffCleanupFn;
    pFileState->stateBuffRemoveFn = stateHashBuffRemoveFn;
    pFileState->stateBuffRemoveByPosFn = stateHashBuffRemoveByPosFn;
    pFileState->stateBuffCreateStateKeyFn = intervalCreateStateKey;

    pFileState->stateFileRemoveFn = intervalFileRemoveFn;
    pFileState->stateFileGetFn = intervalFileGetFn;
    pFileState->cfName = taosStrdup("state");
    pFileState->stateFunctionGetFn = addRowBuffIfNotExist;
  } else if (type == STREAM_STATE_BUFF_SORT) {
    pFileState->rowStateBuff = tSimpleHashInit(cap, hashFn);
    pFileState->stateBuffCleanupFn = sessionWinStateCleanup;
    pFileState->stateBuffRemoveFn = deleteSessionWinStateBuffFn;
    pFileState->stateBuffRemoveByPosFn = deleteSessionWinStateBuffByPosFn;
    pFileState->stateBuffCreateStateKeyFn = sessionCreateStateKey;

    pFileState->stateFileRemoveFn = sessionFileRemoveFn;
    pFileState->stateFileGetFn = sessionFileGetFn;
    pFileState->cfName = taosStrdup("sess");
    pFileState->stateFunctionGetFn = getSessionRowBuff;
  } else if (type == STREAM_STATE_BUFF_HASH_SORT) {
    pFileState->rowStateBuff = tSimpleHashInit(cap, hashFn);
    pFileState->searchBuff = tSimpleHashInit(MIN_NUM_SEARCH_BUCKET, hashFn);
    QUERY_CHECK_NULL(pFileState->searchBuff, code, lino, _end, terrno);
    pFileState->stateBuffCleanupFn = stateHashBuffCleanupFn;
    pFileState->stateBuffRemoveFn = stateHashBuffRemoveFn;
    pFileState->stateBuffRemoveByPosFn = stateHashBuffRemoveByPosFn;
    pFileState->stateBuffCreateStateKeyFn = defaultCreateStateKey;

    pFileState->stateFileRemoveFn = hashSortFileRemoveFn;
    pFileState->stateFileGetFn = hashSortFileGetFn;
    pFileState->cfName = taosStrdup("fill");
    pFileState->stateFunctionGetFn = NULL;
  }

  QUERY_CHECK_NULL(pFileState->usedBuffs, code, lino, _end, terrno);
  QUERY_CHECK_NULL(pFileState->freeBuffs, code, lino, _end, terrno);
  QUERY_CHECK_NULL(pFileState->rowStateBuff, code, lino, _end, terrno);
  QUERY_CHECK_NULL(pFileState->cfName, code, lino, _end, terrno);

  if (type == STREAM_STATE_BUFF_HASH_SEARCH) {
    pFileState->searchBuff = tSimpleHashInit(MIN_NUM_SEARCH_BUCKET, hashFn);
    QUERY_CHECK_NULL(pFileState->searchBuff, code, lino, _end, terrno);
  }

  pFileState->keyLen = keySize;
  pFileState->rowSize = rowSize;
  pFileState->selectivityRowSize = selectRowSize;
  pFileState->preCheckPointVersion = 0;
  pFileState->checkPointVersion = 1;
  pFileState->pFileStore = pFile;
  pFileState->getTs = fp;
  pFileState->curRowCount = 0;
  pFileState->deleteMark = delMark;
  pFileState->flushMark = INT64_MIN;
  pFileState->maxTs = INT64_MIN;
  pFileState->id = taosStrdup(taskId);
  QUERY_CHECK_NULL(pFileState->id, code, lino, _end, terrno);

  pFileState->pGroupIdMap = tSimpleHashInit(1024, hashFn);
  QUERY_CHECK_NULL(pFileState->pGroupIdMap, code, lino, _end, terrno);

  pFileState->pRecFlagMap = tSimpleHashInit(1024, hashFn);
  QUERY_CHECK_NULL(pFileState->pRecFlagMap, code, lino, _end, terrno);


  pFileState->hasFillCatch = true;

  if (type == STREAM_STATE_BUFF_HASH || type == STREAM_STATE_BUFF_HASH_SEARCH) {
    code = recoverSnapshot(pFileState, checkpointId);
  } else if (type == STREAM_STATE_BUFF_SORT) {
    code = recoverSession(pFileState, checkpointId);
  } else if (type == STREAM_STATE_BUFF_HASH_SORT) {
    code = recoverFillSnapshot(pFileState, checkpointId);
  }
  QUERY_CHECK_CODE(code, lino, _end);

  void*   valBuf = NULL;
  int32_t len = 0;
  int32_t tmpRes = streamDefaultGet_rocksdb(pFileState->pFileStore, STREAM_STATE_INFO_NAME, &valBuf, &len);
  if (tmpRes == TSDB_CODE_SUCCESS) {
    QUERY_CHECK_CONDITION((len == sizeof(TSKEY)), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
    streamFileStateDecode(&pFileState->flushMark, valBuf, len);
    qDebug("===stream===flushMark  read:%" PRId64, pFileState->flushMark);
  }
  taosMemoryFreeClear(valBuf);
  (*ppFileState) = pFileState;

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

void destroyRowBuffPos(SRowBuffPos* pPos) {
  taosMemoryFreeClear(pPos->pKey);
  taosMemoryFreeClear(pPos->pRowBuff);
  taosMemoryFree(pPos);
}

void destroyRowBuffPosPtr(void* ptr) {
  if (!ptr) {
    return;
  }
  SRowBuffPos* pPos = *(SRowBuffPos**)ptr;
  if (!pPos->beUsed) {
    destroyRowBuffPos(pPos);
  }
}

void destroyRowBuffAllPosPtr(void* ptr) {
  if (!ptr) {
    return;
  }
  SRowBuffPos* pPos = *(SRowBuffPos**)ptr;
  destroyRowBuffPos(pPos);
}

void destroyRowBuff(void* ptr) {
  if (!ptr) {
    return;
  }
  taosMemoryFree(*(void**)ptr);
}

void streamFileStateDestroy(SStreamFileState* pFileState) {
  if (!pFileState) {
    return;
  }

  taosMemoryFree(pFileState->id);
  taosMemoryFree(pFileState->cfName);
  tdListFreeP(pFileState->usedBuffs, destroyRowBuffAllPosPtr);
  tdListFreeP(pFileState->freeBuffs, destroyRowBuff);
  pFileState->stateBuffCleanupFn(pFileState->rowStateBuff);
  sessionWinStateCleanup(pFileState->searchBuff);
  tSimpleHashCleanup(pFileState->pGroupIdMap);
  tSimpleHashCleanup(pFileState->pRecFlagMap);
  taosMemoryFree(pFileState);
}

int32_t putFreeBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (pPos->pRowBuff) {
    code = tdListAppend(pFileState->freeBuffs, &(pPos->pRowBuff));
    QUERY_CHECK_CODE(code, lino, _end);
    pPos->pRowBuff = NULL;
  }

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

void clearExpiredRowBuff(SStreamFileState* pFileState, TSKEY ts, bool all) {
  int32_t   code = TSDB_CODE_SUCCESS;
  int32_t   lino = 0;
  SListIter iter = {0};
  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&iter)) != NULL) {
    SRowBuffPos* pPos = *(SRowBuffPos**)(pNode->data);
    if (all || (pFileState->getTs(pPos->pKey) < ts && !pPos->beUsed)) {
      code = putFreeBuff(pFileState, pPos);
      QUERY_CHECK_CODE(code, lino, _end);

      if (!all) {
        pFileState->stateBuffRemoveByPosFn(pFileState, pPos);
      }
      destroyRowBuffPos(pPos);
      SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);
      taosMemoryFreeClear(tmp);
    }
  }

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

int32_t clearFlushedRowBuff(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uint64_t max, bool all) {
  int32_t   code = TSDB_CODE_SUCCESS;
  int32_t   lino = 0;
  uint64_t  i = 0;
  SListIter iter = {0};
  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&iter)) != NULL && i < max) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
    if (isFlushedState(pFileState, pFileState->getTs(pPos->pKey), 0)) {
      if (all || !pPos->beUsed) {
        if (all && !pPos->pRowBuff) {
          continue;
        }
        code = tdListAppend(pFlushList, &pPos);
        QUERY_CHECK_CODE(code, lino, _end);

        pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));
        pFileState->stateBuffRemoveByPosFn(pFileState, pPos);
        if (pPos->beUsed == false) {
          SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);
          taosMemoryFreeClear(tmp);
        }
        if (pPos->pRowBuff) {
          i++;
        }
      }
    }
  }
  qDebug("clear flushed row buff. %d rows to disk. is all:%d", listNEles(pFlushList), all);

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

void streamFileStateClear(SStreamFileState* pFileState) {
  pFileState->flushMark = INT64_MIN;
  pFileState->maxTs = INT64_MIN;
  tSimpleHashClear(pFileState->rowStateBuff);
  clearExpiredRowBuff(pFileState, 0, true);
}

bool needClearDiskBuff(SStreamFileState* pFileState) { return pFileState->flushMark > 0; }

void streamFileStateReleaseBuff(SStreamFileState* pFileState, SRowBuffPos* pPos, bool used) { pPos->beUsed = used; }

int32_t popUsedBuffs(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uint64_t max, bool used) {
  int32_t   code = TSDB_CODE_SUCCESS;
  int32_t   lino = 0;
  uint64_t  i = 0;
  SListIter iter = {0};
  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&iter)) != NULL && i < max) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
    if (pPos->beUsed == used) {
      if (used && !pPos->pRowBuff) {
        continue;
      }
      code = tdListAppend(pFlushList, &pPos);
      QUERY_CHECK_CODE(code, lino, _end);

      pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));
      pFileState->stateBuffRemoveByPosFn(pFileState, pPos);
      if (pPos->beUsed == false) {
        SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);
        taosMemoryFreeClear(tmp);
      }
      if (pPos->pRowBuff) {
        i++;
      }
    }
  }

  qInfo("%s stream state flush %d rows to disk. is used:%d", pFileState->id, listNEles(pFlushList), used);

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

int32_t flushRowBuff(SStreamFileState* pFileState) {
  int32_t          code = TSDB_CODE_SUCCESS;
  int32_t          lino = 0;
  SStreamSnapshot* pFlushList = tdListNew(POINTER_BYTES);
  if (!pFlushList) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  uint64_t num = (uint64_t)(pFileState->curRowCount * FLUSH_RATIO);
  num = TMAX(num, FLUSH_NUM);
  code = clearFlushedRowBuff(pFileState, pFlushList, num, false);
  QUERY_CHECK_CODE(code, lino, _end);

  if (isListEmpty(pFlushList)) {
    code = popUsedBuffs(pFileState, pFlushList, num, false);
    QUERY_CHECK_CODE(code, lino, _end);

    if (isListEmpty(pFlushList)) {
      code = popUsedBuffs(pFileState, pFlushList, num, true);
      QUERY_CHECK_CODE(code, lino, _end);
    }
  }

  if (pFileState->searchBuff) {
    code = clearFlushedRowBuff(pFileState, pFlushList, pFileState->curRowCount, true);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  flushSnapshot(pFileState, pFlushList, false);

  SListIter fIter = {0};
  tdListInitIter(pFlushList, &fIter, TD_LIST_FORWARD);
  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&fIter)) != NULL) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
    code = putFreeBuff(pFileState, pPos);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  tdListFreeP(pFlushList, destroyRowBuffPosPtr);

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

int32_t clearRowBuff(SStreamFileState* pFileState) {
  if (pFileState->deleteMark != INT64_MAX) {
    clearExpiredRowBuff(pFileState, pFileState->maxTs - pFileState->deleteMark, false);
  }
  do {
    int32_t code = flushRowBuff(pFileState);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  } while (isListEmpty(pFileState->freeBuffs) && pFileState->curRowCount == pFileState->maxRowCount);
  return TSDB_CODE_SUCCESS;
}

int32_t clearFlushedRowBuffByFlag(SStreamFileState* pFileState, uint64_t max) {
  int32_t   code = TSDB_CODE_SUCCESS;
  int32_t   lino = 0;
  uint64_t  i = 0;
  SListIter iter = {0};
  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&iter)) != NULL && i < max) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
    if (pPos->invalid) {
      if (!pPos->beUsed) {
        SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);
        taosMemoryFreeClear(tmp);
        if (pPos->pRowBuff) {
          i++;
        }
        code = putFreeBuff(pFileState, pPos);
        QUERY_CHECK_CODE(code, lino, _end);
        destroyRowBuffPos(pPos);
      }
    }
  }

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

int32_t clearRowBuffNonFlush(SStreamFileState* pFileState) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  if (pFileState->deleteMark != INT64_MAX) {
    clearExpiredRowBuff(pFileState, pFileState->maxTs - pFileState->deleteMark, false);
  }

  uint64_t num = (uint64_t)(pFileState->curRowCount * FLUSH_RATIO);
  num = TMAX(num, FLUSH_NUM);
  code = clearFlushedRowBuffByFlag(pFileState, num);
  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* getFreeBuff(SStreamFileState* pFileState) {
  SList*     lists = pFileState->freeBuffs;
  int32_t    buffSize = pFileState->rowSize;
  SListNode* pNode = tdListPopHead(lists);
  if (!pNode) {
    return NULL;
  }
  void* ptr = *(void**)pNode->data;
  memset(ptr, 0, buffSize);
  taosMemoryFree(pNode);
  return ptr;
}

void streamFileStateClearBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {
  if (pPos->pRowBuff) {
    memset(pPos->pRowBuff, 0, pFileState->rowSize);
  }
}

SRowBuffPos* getNewRowPos(SStreamFileState* pFileState) {
  int32_t      code = TSDB_CODE_SUCCESS;
  int32_t      lino = 0;
  SRowBuffPos* pPos = taosMemoryCalloc(1, sizeof(SRowBuffPos));
  if (!pPos) {
    code = terrno;
    QUERY_CHECK_CODE(code, lino, _error);
  }

  pPos->pKey = taosMemoryCalloc(1, pFileState->keyLen);
  if (!pPos->pKey) {
    code = terrno;
    QUERY_CHECK_CODE(code, lino, _error);
  }

  void* pBuff = getFreeBuff(pFileState);
  if (pBuff) {
    pPos->pRowBuff = pBuff;
    goto _end;
  }

  if (pFileState->curRowCount < pFileState->maxRowCount) {
    pBuff = taosMemoryCalloc(1, pFileState->rowSize);
    QUERY_CHECK_NULL(pBuff, code, lino, _error, terrno);
    pPos->pRowBuff = pBuff;
    pFileState->curRowCount++;
    goto _end;
  }

  code = clearRowBuff(pFileState);
  QUERY_CHECK_CODE(code, lino, _error);

  pPos->pRowBuff = getFreeBuff(pFileState);

_end:
  code = tdListAppend(pFileState->usedBuffs, &pPos);
  QUERY_CHECK_CODE(code, lino, _error);

  QUERY_CHECK_CONDITION((pPos->pRowBuff != NULL), code, lino, _error, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
_error:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
    return NULL;
  }

  return pPos;
}

SRowBuffPos* getNewRowPosForWrite(SStreamFileState* pFileState) {
  int32_t      code = TSDB_CODE_SUCCESS;
  int32_t      lino = 0;
  SRowBuffPos* newPos = getNewRowPos(pFileState);
  if (!newPos) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    QUERY_CHECK_CODE(code, lino, _error);
  }
  newPos->beUsed = true;
  newPos->beFlushed = false;
  newPos->needFree = false;
  newPos->beUpdated = true;
  newPos->invalid = false;
  return newPos;

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

int32_t addRowBuffIfNotExist(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen,
                             int32_t* pWinCode) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  (*pWinCode) = TSDB_CODE_SUCCESS;
  pFileState->maxTs = TMAX(pFileState->maxTs, pFileState->getTs(pKey));
  SRowBuffPos** pos = tSimpleHashGet(pFileState->rowStateBuff, pKey, keyLen);
  if (pos) {
    if (pVal != NULL) {
      *pVLen = pFileState->rowSize;
      *pVal = *pos;
      (*pos)->beUsed = true;
      (*pos)->beFlushed = false;
    }
    goto _end;
  }
  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
  if (!pNewPos || !pNewPos->pRowBuff) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  memcpy(pNewPos->pKey, pKey, keyLen);
  (*pWinCode) = TSDB_CODE_FAILED;

  TSKEY ts = pFileState->getTs(pKey);
  if (!isDeteled(pFileState, ts) && isFlushedState(pFileState, ts, 0)) {
    int32_t len = 0;
    void*   p = NULL;
    (*pWinCode) = pFileState->stateFileGetFn(pFileState, pKey, &p, &len);
    qDebug("===stream===get %" PRId64 " from disc, res %d", ts, (*pWinCode));
    if ((*pWinCode) == TSDB_CODE_SUCCESS) {
      memcpy(pNewPos->pRowBuff, p, len);
    }
    taosMemoryFree(p);
  }

  code = tSimpleHashPut(pFileState->rowStateBuff, pKey, keyLen, &pNewPos, POINTER_BYTES);
  QUERY_CHECK_CODE(code, lino, _end);

  if (pVal) {
    *pVLen = pFileState->rowSize;
    *pVal = pNewPos;
  }

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

int32_t createRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  pFileState->maxTs = TMAX(pFileState->maxTs, pFileState->getTs(pKey));

  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
  if (!pNewPos || !pNewPos->pRowBuff) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  memcpy(pNewPos->pKey, pKey, keyLen);

  code = tSimpleHashPut(pFileState->rowStateBuff, pKey, keyLen, &pNewPos, POINTER_BYTES);
  QUERY_CHECK_CODE(code, lino, _end);

  if (pVal) {
    *pVLen = pFileState->rowSize;
    *pVal = pNewPos;
  }

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

void deleteRowBuff(SStreamFileState* pFileState, const void* pKey, int32_t keyLen) {
  int32_t code_buff = pFileState->stateBuffRemoveFn(pFileState->rowStateBuff, pKey, keyLen);
  qTrace("%s at line %d res:%d", __func__, __LINE__, code_buff);
  int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey);
  qTrace("%s at line %d res:%d", __func__, __LINE__, code_file);
  if (pFileState->searchBuff != NULL) {
    deleteHashSortRowBuff(pFileState, pKey);
  }
}

void deleteRowBuffByGroupId(SStreamFileState* pFileState, uint64_t groupId) {
  SSHashObj* pRowMap = pFileState->rowStateBuff;
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pRowMap, pIte, &iter)) != NULL) {
    size_t keyLen = 0;
    SWinKey* pKey = tSimpleHashGetKey(pIte, &keyLen);
    if (pKey->groupId == groupId) {
      int32_t tmpRes = tSimpleHashIterateRemove(pRowMap, pKey, keyLen, &pIte, &iter);
      qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
    }
  }

  while (1) {
    SWinKey tmp = {.ts = INT64_MIN, .groupId = groupId};
    SStreamStateCur* pCur = streamStateSeekKeyNext_rocksdb(pFileState->pFileStore, &tmp);
    SWinKey delKey = {.groupId = groupId};
    int32_t code = streamStateGetGroupKVByCur_rocksdb(pFileState->pFileStore, pCur, &delKey, NULL, 0);
    if (code != TSDB_CODE_SUCCESS) {
      break;
    }
    code = streamStateDel_rocksdb(pFileState->pFileStore, &delKey);
    qTrace("%s at line %d res:%d", __func__, __LINE__, code);
  }
}

static int32_t recoverSessionRowBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  int32_t len = 0;
  void*   pBuff = NULL;
  code = pFileState->stateFileGetFn(pFileState, pPos->pKey, &pBuff, &len);
  QUERY_CHECK_CODE(code, lino, _end);
  memcpy(pPos->pRowBuff, pBuff, len);
  taosMemoryFree(pBuff);

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

static int32_t recoverStateRowBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  pPos->pRowBuff = getFreeBuff(pFileState);
  if (!pPos->pRowBuff) {
    if (pFileState->curRowCount < pFileState->maxRowCount) {
      pPos->pRowBuff = taosMemoryCalloc(1, pFileState->rowSize);
      if (!pPos->pRowBuff) {
        code = terrno;
        QUERY_CHECK_CODE(code, lino, _end);
      }
      pFileState->curRowCount++;
    } else {
      code = clearRowBuff(pFileState);
      QUERY_CHECK_CODE(code, lino, _end);
      pPos->pRowBuff = getFreeBuff(pFileState);
    }
    QUERY_CHECK_CONDITION((pPos->pRowBuff != NULL), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);
  }

  code = recoverSessionRowBuff(pFileState, pPos);
  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 getRowBuffByPos(SStreamFileState* pFileState, SRowBuffPos* pPos, void** pVal) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (pPos->pRowBuff) {
    if (pPos->needFree) {
      code = recoverSessionRowBuff(pFileState, pPos);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    (*pVal) = pPos->pRowBuff;
    goto _end;
  }

  code = recoverStateRowBuff(pFileState, pPos);
  QUERY_CHECK_CODE(code, lino, _end);

  (*pVal) = pPos->pRowBuff;
  // if (!pPos->needFree) {
  //   code = tdListPrepend(pFileState->usedBuffs, &pPos);
  //   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;
}

bool hasRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, bool hasLimit, bool* pIsLast) {
  bool res = false;
  if (pIsLast != NULL) {
    (*pIsLast) = false;
  }
  
  SRowBuffPos** pos = tSimpleHashGet(pFileState->rowStateBuff, pKey, sizeof(SWinKey));
  if (pos) {
    res = true;
  }
  void* pSearchBuff = getSearchBuff(pFileState);
  if (pSearchBuff != NULL) {
    void** ppBuff = (void**)tSimpleHashGet(pSearchBuff, &pKey->groupId, sizeof(uint64_t));
    if (ppBuff != NULL) {
      SArray* pWinStates = (SArray*)(*ppBuff);
      if (pIsLast != NULL) {
        SWinKey* pLastKey = (SWinKey*) taosArrayGetLast(pWinStates);
        *pIsLast = (winKeyCmprImpl(pKey, pLastKey) == 0);
      }
      if (hasLimit && taosArrayGetSize(pWinStates) <= MIN_NUM_OF_SORT_CACHE_WIN) {
        res = true;
      }
      if (qDebugFlag & DEBUG_DEBUG) {
        SWinKey* fistKey = (SWinKey*)taosArrayGet(pWinStates, 0);
        qDebug("===stream===check window state. buff min ts:%" PRId64 ",groupId:%" PRIu64 ".key ts:%" PRId64
               ",groupId:%" PRIu64,
               fistKey->ts, fistKey->groupId, pKey->ts, pKey->groupId);
      }
    } else {
      res = true;
    }
  }
  return res;
}

SStreamSnapshot* getSnapshot(SStreamFileState* pFileState) {
  int64_t mark = (pFileState->deleteMark == INT64_MAX || pFileState->maxTs == INT64_MIN)
                     ? INT64_MIN
                     : pFileState->maxTs - pFileState->deleteMark;
  clearExpiredRowBuff(pFileState, mark, false);
  return pFileState->usedBuffs;
}

void flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState) {
  int32_t   code = TSDB_CODE_SUCCESS;
  int32_t   lino = 0;
  SListIter iter = {0};
  tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD);

  int64_t    st = taosGetTimestampMs();
  SListNode* pNode = NULL;

  int idx = streamStateGetCfIdx(pFileState->pFileStore, pFileState->cfName);

  int32_t len = (pFileState->rowSize + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;
  char*   buf = taosMemoryCalloc(1, len);
  if (!buf) {
    code = terrno;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  void* batch = streamStateCreateBatch();
  if (!batch) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  while ((pNode = tdListNext(&iter)) != NULL && code == TSDB_CODE_SUCCESS) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
    if (pPos->beFlushed || !pPos->pRowBuff) {
      continue;
    }
    pPos->beFlushed = true;
    pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));

    qDebug("===stream===flushed start:%" PRId64, pFileState->getTs(pPos->pKey));
    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {
      code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
      streamStateClearBatch(batch);
      QUERY_CHECK_CODE(code, lino, _end);
    }

    void* pSKey = pFileState->stateBuffCreateStateKeyFn(pPos, ((SStreamState*)pFileState->pFileStore)->number);
    QUERY_CHECK_NULL(pSKey, code, lino, _end, terrno);

    code = streamStatePutBatchOptimize(pFileState->pFileStore, idx, batch, pSKey, pPos->pRowBuff, pFileState->rowSize,
                                       0, buf);
    taosMemoryFreeClear(pSKey);
    QUERY_CHECK_CODE(code, lino, _end);
    // todo handle failure
    memset(buf, 0, len);
  }
  taosMemoryFreeClear(buf);

  int32_t numOfElems = streamStateGetBatchSize(batch);
  if (numOfElems > 0) {
    code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
    QUERY_CHECK_CODE(code, lino, _end);
  } else {
    goto _end;
  }

  streamStateClearBatch(batch);

  clearSearchBuff(pFileState);

  int64_t elapsed = taosGetTimestampMs() - st;
  qDebug("%s flush to disk in batch model completed, rows:%d, batch size:%d, elapsed time:%" PRId64 "ms",
         pFileState->id, numOfElems, BATCH_LIMIT, elapsed);

  if (flushState) {
    void*   valBuf = NULL;
    int32_t len = 0;
    code = streamFileStateEncode(&pFileState->flushMark, &valBuf, &len);
    QUERY_CHECK_CODE(code, lino, _end);

    qDebug("===stream===flushMark write:%" PRId64, pFileState->flushMark);
    code = streamStatePutBatch(pFileState->pFileStore, "default", batch, STREAM_STATE_INFO_NAME, valBuf, len, 0);
    taosMemoryFree(valBuf);
    QUERY_CHECK_CODE(code, lino, _end);

    code = streamStatePutBatch_rocksdb(pFileState->pFileStore, 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));
  }
  taosMemoryFree(buf);
  streamStateDestroyBatch(batch);
}

int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {
  char keyBuf[128] = {0};
  TAOS_UNUSED(tsnprintf(keyBuf, sizeof(keyBuf), "%s:%" PRId64 "", TASK_KEY, checkpointId));
  return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);
}

int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
  int32_t code = TSDB_CODE_SUCCESS;
  int64_t maxCheckPointId = 0;
  {
    char    buf[128] = {0};
    void*   val = NULL;
    int32_t len = 0;
    memcpy(buf, TASK_KEY, strlen(TASK_KEY));
    code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
    if (code != 0 || len == 0 || val == NULL) {
      return TSDB_CODE_FAILED;
    }
    memcpy(buf, val, len);
    buf[len] = 0;
    maxCheckPointId = taosStr2Int64((char*)buf, NULL, 10);
    taosMemoryFree(val);
  }
  for (int64_t i = maxCheckPointId; i > 0; i--) {
    char    buf[128] = {0};
    void*   val = 0;
    int32_t len = 0;
    TAOS_UNUSED(tsnprintf(buf, sizeof(buf), "%s:%" PRId64 "", TASK_KEY, i));
    code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
    if (code != 0) {
      return TSDB_CODE_FAILED;
    }
    memcpy(buf, val, len);
    buf[len] = 0;
    taosMemoryFree(val);

    TSKEY ts;
    ts = taosStr2Int64((char*)buf, NULL, 10);
    if (ts < mark) {
      // statekey winkey.ts < mark
      int32_t tmpRes = forceRemoveCheckpoint(pFileState, i);
      qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
      break;
    }
  }
  return code;
}

int32_t recoverSession(SStreamFileState* pFileState, int64_t ckId) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  int32_t winRes = TSDB_CODE_SUCCESS;
  if (pFileState->maxTs != INT64_MIN) {
    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)
                       ? INT64_MIN
                       : pFileState->maxTs - pFileState->deleteMark;
    int32_t tmpRes = deleteExpiredCheckPoint(pFileState, mark);
    qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
  }

  SStreamStateCur* pCur = streamStateSessionSeekToLast_rocksdb(pFileState->pFileStore, INT64_MAX);
  int32_t          recoverNum = TMIN(MIN_NUM_OF_RECOVER_ROW_BUFF, pFileState->maxRowCount);
  while (winRes == TSDB_CODE_SUCCESS) {
    if (pFileState->curRowCount >= recoverNum) {
      break;
    }

    void*       pVal = NULL;
    int32_t     vlen = 0;
    SSessionKey key = {0};
    winRes = streamStateSessionGetKVByCur_rocksdb(getStateFileStore(pFileState), pCur, &key, &pVal, &vlen);
    if (winRes != TSDB_CODE_SUCCESS) {
      break;
    }

    if (vlen != pFileState->rowSize) {
      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
      QUERY_CHECK_CODE(code, lino, _end);
    }

    SRowBuffPos* pPos = createSessionWinBuff(pFileState, &key, pVal, &vlen);
    pPos->beUsed = false;
    winRes = putSessionWinResultBuff(pFileState, pPos);
    if (winRes != TSDB_CODE_SUCCESS) {
      break;
    }

    winRes = streamStateSessionCurPrev_rocksdb(pCur);
  }

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

int32_t recoverSnapshot(SStreamFileState* pFileState, int64_t ckId) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  int32_t winCode = TSDB_CODE_SUCCESS;
  if (pFileState->maxTs != INT64_MIN) {
    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)
                       ? INT64_MIN
                       : pFileState->maxTs - pFileState->deleteMark;
    int32_t tmpRes = deleteExpiredCheckPoint(pFileState, mark);
    qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);
  }

  SStreamStateCur* pCur = streamStateSeekToLast_rocksdb(pFileState->pFileStore);
  int32_t          recoverNum = TMIN(MIN_NUM_OF_RECOVER_ROW_BUFF, pFileState->maxRowCount);
  while (winCode == TSDB_CODE_SUCCESS) {
    if (pFileState->curRowCount >= recoverNum) {
      break;
    }

    void*        pVal = NULL;
    int32_t      vlen = 0;
    SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
    if (!pNewPos || !pNewPos->pRowBuff) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      QUERY_CHECK_CODE(code, lino, _end);
    }

    winCode =
        streamStateGetKVByCur_rocksdb(getStateFileStore(pFileState), pCur, pNewPos->pKey, (const void**)&pVal, &vlen);
    qDebug("===stream=== get state by cur winres:%d. %s", winCode, __func__);
    if (winCode != TSDB_CODE_SUCCESS || pFileState->getTs(pNewPos->pKey) < pFileState->flushMark) {
      destroyRowBuffPos(pNewPos);
      SListNode* pNode = tdListPopTail(pFileState->usedBuffs);
      taosMemoryFreeClear(pNode);
      taosMemoryFreeClear(pVal);
      break;
    }
    if (vlen != pFileState->rowSize) {
      qError("row size mismatch, expect:%d, actual:%d", pFileState->rowSize, vlen);
      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
      taosMemoryFreeClear(pVal);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    memcpy(pNewPos->pRowBuff, pVal, vlen);
    taosMemoryFreeClear(pVal);
    pNewPos->beFlushed = true;
    pNewPos->beUsed = false;
    qDebug("===stream=== read checkpoint state from disc. %s", __func__);
    code = tSimpleHashPut(pFileState->rowStateBuff, pNewPos->pKey, pFileState->keyLen, &pNewPos, POINTER_BYTES);
    if (code != TSDB_CODE_SUCCESS) {
      destroyRowBuffPos(pNewPos);
      break;
    }
    streamStateCurPrev_rocksdb(pCur);
  }

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

int32_t streamFileStateGetSelectRowSize(SStreamFileState* pFileState) { return pFileState->selectivityRowSize; }

void streamFileStateReloadInfo(SStreamFileState* pFileState, TSKEY ts) {
  pFileState->flushMark = TMAX(pFileState->flushMark, ts);
  pFileState->maxTs = TMAX(pFileState->maxTs, ts);
}

void* getRowStateBuff(SStreamFileState* pFileState) { return pFileState->rowStateBuff; }
void* getSearchBuff(SStreamFileState* pFileState) { return pFileState->searchBuff; }

void* getStateFileStore(SStreamFileState* pFileState) { return pFileState->pFileStore; }

bool isDeteled(SStreamFileState* pFileState, TSKEY ts) {
  return pFileState->deleteMark != INT64_MAX && pFileState->maxTs > 0 &&
         ts < (pFileState->maxTs - pFileState->deleteMark);
}

bool isFlushedState(SStreamFileState* pFileState, TSKEY ts, TSKEY gap) { return ts <= (pFileState->flushMark + gap); }

TSKEY getFlushMark(SStreamFileState* pFileState) { return pFileState->flushMark; };

int32_t getRowStateRowSize(SStreamFileState* pFileState) { return pFileState->rowSize; }

int32_t getFunctionRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen) {
  int32_t winCode = TSDB_CODE_SUCCESS;
  return pFileState->stateFunctionGetFn(pFileState, pKey, keyLen, pVal, pVLen, &winCode);
}

int32_t recoverFillSnapshot(SStreamFileState* pFileState, int64_t ckId) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (pFileState->maxTs != INT64_MIN) {
    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)
                       ? INT64_MIN
                       : pFileState->maxTs - pFileState->deleteMark;
    code = deleteExpiredCheckPoint(pFileState, mark);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  SStreamStateCur* pCur = streamStateFillSeekToLast_rocksdb(pFileState->pFileStore);
  if (pCur == NULL) {
    return code;
  }
  int32_t recoverNum = TMIN(MIN_NUM_OF_RECOVER_ROW_BUFF, pFileState->maxRowCount);
  int32_t winRes = TSDB_CODE_SUCCESS;
  while (winRes == TSDB_CODE_SUCCESS) {
    if (pFileState->curRowCount >= recoverNum) {
      break;
    }

    void*        pVal = NULL;
    int32_t      vlen = 0;
    SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
    winRes = streamStateFillGetKVByCur_rocksdb(pCur, pNewPos->pKey, (const void**)&pVal, &vlen);
    qDebug("===stream=== get state by cur winres:%d. %s", winRes, __func__);
    if (winRes != TSDB_CODE_SUCCESS || isFlushedState(pFileState, pFileState->getTs(pNewPos->pKey), 0)) {
      destroyRowBuffPos(pNewPos);
      SListNode* pNode = tdListPopTail(pFileState->usedBuffs);
      taosMemoryFreeClear(pNode);
      taosMemoryFreeClear(pVal);
      break;
    }

    if (vlen != pFileState->rowSize) {
      qError("row size mismatch, expect:%d, actual:%d", pFileState->rowSize, vlen);
      destroyRowBuffPos(pNewPos);
      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
      taosMemoryFreeClear(pVal);
      QUERY_CHECK_CODE(code, lino, _end);
    }

    memcpy(pNewPos->pRowBuff, pVal, vlen);
    taosMemoryFreeClear(pVal);
    pNewPos->beFlushed = true;
    pNewPos->beUsed = false;
    qDebug("===stream=== read checkpoint state from disc. %s", __func__);
    winRes = tSimpleHashPut(pFileState->rowStateBuff, pNewPos->pKey, pFileState->keyLen, &pNewPos, POINTER_BYTES);
    if (winRes != TSDB_CODE_SUCCESS) {
      destroyRowBuffPos(pNewPos);
      break;
    }
    streamStateCurPrev_rocksdb(pCur);
  }
  streamStateFreeCur(pCur);

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

int32_t getRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen,
                   int32_t* pWinCode) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  (*pWinCode) = TSDB_CODE_FAILED;
  pFileState->maxTs = TMAX(pFileState->maxTs, pFileState->getTs(pKey));
  SRowBuffPos** ppPos = tSimpleHashGet(pFileState->rowStateBuff, pKey, keyLen);
  if (ppPos) {
    *pVLen = pFileState->rowSize;
    *pVal = *ppPos;
    (*ppPos)->beUsed = true;
    (*ppPos)->beFlushed = false;
    (*pWinCode) = TSDB_CODE_SUCCESS;
    if ((*ppPos)->pRowBuff == NULL) {
      code = recoverStateRowBuff(pFileState, *ppPos);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    goto _end;
  }
  TSKEY ts = pFileState->getTs(pKey);
  if (!isDeteled(pFileState, ts) && isFlushedState(pFileState, ts, 0)) {
    int32_t len = 0;
    void*   p = NULL;
    (*pWinCode) = pFileState->stateFileGetFn(pFileState, pKey, &p, &len);
    qDebug("===stream===get %" PRId64 " from disc, res %d", ts, (*pWinCode));
    if ((*pWinCode) == TSDB_CODE_SUCCESS) {
      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
      if (!pNewPos || !pNewPos->pRowBuff) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        QUERY_CHECK_CODE(code, lino, _end);
      }

      memcpy(pNewPos->pKey, pKey, keyLen);
      memcpy(pNewPos->pRowBuff, p, len);
      code = tSimpleHashPut(pFileState->rowStateBuff, pKey, keyLen, &pNewPos, POINTER_BYTES);
      QUERY_CHECK_CODE(code, lino, _end);

      if (pVal) {
        *pVLen = pFileState->rowSize;
        *pVal = pNewPos;
      }
    }
    taosMemoryFree(p);
  }

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

int32_t streamFileStateGroupPut(SStreamFileState* pFileState, int64_t groupId, void* value, int32_t vLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  if (value != NULL) {
    code = TSDB_CODE_INVALID_PARA;
    QUERY_CHECK_CODE(code, lino, _end);
  }

  if (tSimpleHashGet(pFileState->pGroupIdMap, &groupId, sizeof(int64_t)) == NULL) {
    if (tSimpleHashGetSize(pFileState->pGroupIdMap) <= MAX_GROUP_ID_NUM) {
      code = tSimpleHashPut(pFileState->pGroupIdMap, &groupId, sizeof(int64_t), NULL, 0);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    code = streamStatePutParTag_rocksdb(pFileState->pFileStore, groupId, value, vLen);
    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 streamFileStateGroupCurNext(SStreamStateCur* pCur) {
  SStreamFileState* pFileState = (SStreamFileState*)pCur->pStreamFileState;
  if (pCur->hashIter == -1) {
    streamStateCurNext(pFileState->pFileStore, pCur);
    return;
  }

  int64_t gpId = *(int64_t*)tSimpleHashGetKey(pCur->pHashData, NULL);
  pCur->minGpId = TMAX(pCur->minGpId, gpId);

  SSHashObj* pHash = pFileState->pGroupIdMap;
  pCur->pHashData = tSimpleHashIterate(pHash, pCur->pHashData, &pCur->hashIter);
  if (!pCur->pHashData) {
    pCur->hashIter = -1;
    streamStateParTagSeekKeyNext_rocksdb(pFileState->pFileStore, pCur->minGpId, pCur);
    return;
  }
}

int32_t streamFileStateGroupGetKVByCur(SStreamStateCur* pCur, int64_t* pKey, void** pVal, int32_t* pVLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  if (pCur->pHashData) {
    *pKey = *(int64_t*)tSimpleHashGetKey(pCur->pHashData, NULL);
    return code;
  }
  return streamStateParTagGetKVByCur_rocksdb(pCur, pKey, NULL, NULL);
}

SSHashObj* getGroupIdCache(SStreamFileState* pFileState) {
  return pFileState->pGroupIdMap;
}

void clearExpiredState(SStreamFileState* pFileState, int32_t numOfKeep, TSKEY minTs) {
  int32_t    code = TSDB_CODE_SUCCESS;
  int32_t    lino = 0;
  SSHashObj* pSearchBuff = pFileState->searchBuff;
  void*      pIte = NULL;
  int32_t    iter = 0;
  while ((pIte = tSimpleHashIterate(pSearchBuff, pIte, &iter)) != NULL) {
    SArray* pWinStates = *((void**)pIte);
    int32_t arraySize = TARRAY_SIZE(pWinStates);
    if (minTs != INT64_MAX && arraySize > numOfKeep) {
      SWinKey key = {.ts = minTs};
      key.groupId = *(uint64_t*)tSimpleHashGetKey(pIte, NULL);
      int32_t index = binarySearch(pWinStates, arraySize, &key, fillStateKeyCompare);
      numOfKeep = TMAX(arraySize - index, MIN_NUM_OF_SORT_CACHE_WIN);
      qDebug("modify numOfKeep, numOfKeep:%d. %s at line %d", numOfKeep, __func__, __LINE__);
    }

    int32_t size = arraySize - numOfKeep;
    for (int32_t i = 0; i < size; i++) {
      SWinKey* pKey = taosArrayGet(pWinStates, i);
      int32_t  code_buff = pFileState->stateBuffRemoveFn(pFileState->rowStateBuff, pKey, sizeof(SWinKey));
      qTrace("clear expired buff, ts:%" PRId64 ",groupid:%" PRIu64 ". %s at line %d res:%d", pKey->ts, pKey->groupId, __func__, __LINE__, code_buff);

      if (isFlushedState(pFileState, pKey->ts, 0)) {
        int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey);
        qTrace("clear expired file, ts:%" PRId64 ". %s at line %d res:%d", pKey->ts, __func__, __LINE__, code_file);
      }

      if (tSimpleHashGetSize(pFileState->pRecFlagMap) > 0) {
        tSimpleHashRemove(pFileState->pRecFlagMap, pKey, sizeof(SWinKey));
      }
    }
    taosArrayRemoveBatch(pWinStates, 0, size, NULL);
  }
  code = clearRowBuffNonFlush(pFileState);
  QUERY_CHECK_CODE(code, lino, _end);

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

#ifdef BUILD_NO_CALL
int32_t getStateSearchRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, void** pVal, int32_t* pVLen,
                              int32_t* pWinCode) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  code = addRowBuffIfNotExist(pFileState, (void*)pKey, sizeof(SWinKey), pVal, pVLen, pWinCode);
  QUERY_CHECK_CODE(code, lino, _end);

  SArray*    pWinStates = NULL;
  SSHashObj* pSearchBuff = getSearchBuff(pFileState);
  void**     ppBuff = tSimpleHashGet(pSearchBuff, &pKey->groupId, sizeof(uint64_t));
  if (ppBuff) {
    pWinStates = (SArray*)(*ppBuff);
  } else {
    pWinStates = taosArrayInit(16, sizeof(SWinKey));
    QUERY_CHECK_NULL(pWinStates, code, lino, _end, terrno);

    code = tSimpleHashPut(pSearchBuff, &pKey->groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  // recover
  if (taosArrayGetSize(pWinStates) == 0 && needClearDiskBuff(pFileState)) {
    recoverHashSortBuff(pFileState, pWinStates, pKey->groupId);
  }

  int32_t size = taosArrayGetSize(pWinStates);
  int32_t index = binarySearch(pWinStates, size, pKey, fillStateKeyCompare);
  if (!isFlushedState(pFileState, pKey->ts, 0) || index >= 0) {
    // find the first position which is smaller than the pKey
    if (index >= 0) {
      SWinKey* pTmpKey = taosArrayGet(pWinStates, index);
      if (winKeyCmprImpl(pTmpKey, pKey) == 0) {
        goto _end;
      }
    }
    index++;
    void* tmp = taosArrayInsert(pWinStates, index, pKey);
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
  }

  if (size >= MAX_NUM_OF_CACHE_WIN) {
    int32_t num = size - NUM_OF_CACHE_WIN;
    taosArrayRemoveBatch(pWinStates, 0, num, NULL);
  }

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

int32_t getRowStatePrevRow(SStreamFileState* pFileState, const SWinKey* pKey, SWinKey* pResKey, void** ppVal,
                           int32_t* pVLen, int32_t* pWinCode) {
  int32_t    code = TSDB_CODE_SUCCESS;
  int32_t    lino = 0;
  SArray*    pWinStates = NULL;
  SSHashObj* pSearchBuff = getSearchBuff(pFileState);
  void*      pState = getStateFileStore(pFileState);
  void**     ppBuff = (void**)tSimpleHashGet(pSearchBuff, &pKey->groupId, sizeof(uint64_t));
  if (ppBuff) {
    pWinStates = (SArray*)(*ppBuff);
  } else if (needClearDiskBuff(pFileState)) {
    qDebug("===stream=== search buff is empty.group id:%" PRId64, pKey->groupId);
    SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, pKey);
    void*            tmpVal = NULL;
    int32_t          len = 0;
    (*pWinCode) = streamStateGetGroupKVByCur_rocksdb(pState, pCur, pResKey, (const void**)&tmpVal, &len);
    if ((*pWinCode) == TSDB_CODE_SUCCESS) {
      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
      if (!pNewPos || !pNewPos->pRowBuff) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        QUERY_CHECK_CODE(code, lino, _end);
      }
      memcpy(pNewPos->pRowBuff, tmpVal, len);
      taosMemoryFreeClear(tmpVal);
      *pVLen = getRowStateRowSize(pFileState);
      (*ppVal) = pNewPos;
    }
    streamStateFreeCur(pCur);
    return code;
  } else {
    (*pWinCode) = TSDB_CODE_FAILED;
    return code;
  }
  int32_t size = taosArrayGetSize(pWinStates);
  int32_t index = binarySearch(pWinStates, size, pKey, fillStateKeyCompare);
  if (index >= 0) {
    SWinKey* pCurKey = taosArrayGet(pWinStates, index);
    if (winKeyCmprImpl(pCurKey, pKey) == 0) {
      index--;
    } else {
      qDebug("%s failed at line %d since do not find cur SWinKey. trigger may be force window close", __func__,
             __LINE__);
    }
  }
  if (index == -1) {
    SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, pKey);
    void*            tmpVal = NULL;
    int32_t          len = 0;
    (*pWinCode) = streamStateGetGroupKVByCur_rocksdb(pState, pCur, pResKey, (const void**)&tmpVal, &len);
    if ((*pWinCode) == TSDB_CODE_SUCCESS) {
      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);
      if (!pNewPos || !pNewPos->pRowBuff) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        QUERY_CHECK_CODE(code, lino, _end);
      }
      memcpy(pNewPos->pRowBuff, tmpVal, len);
      taosMemoryFreeClear(tmpVal);
      *pVLen = getRowStateRowSize(pFileState);
      (*ppVal) = pNewPos;
    }
    streamStateFreeCur(pCur);
    return code;
  } else {
    SWinKey* pPrevKey = taosArrayGet(pWinStates, index);
    *pResKey = *pPrevKey;
    return addRowBuffIfNotExist(pFileState, (void*)pPrevKey, sizeof(SWinKey), ppVal, pVLen, pWinCode);
  }
  (*pWinCode) = TSDB_CODE_FAILED;

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

int32_t addSearchItem(SStreamFileState* pFileState, SArray* pWinStates, const SWinKey* pKey, bool* pIsEnd) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  int32_t size = taosArrayGetSize(pWinStates);
  int32_t index = binarySearch(pWinStates, size, pKey, fillTSKeyCompare);
  if (!isFlushedState(pFileState, pKey->ts, 0) || index >= 0 || size == 0) {
    if (index >= 0) {
      SWinKey* pTmpKey = taosArrayGet(pWinStates, index);
      if (winKeyCmprImpl(pTmpKey, pKey) == 0) {
        goto _end;
      }
    }
    index++;
    (*pIsEnd) = (index >= size);
    void* tmp = taosArrayInsert(pWinStates, index, pKey);
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
  }

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

int32_t addArrayBuffIfNotExist(SSHashObj* pSearchBuff, uint64_t groupId, SArray** ppResStates) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  SArray* pWinStates = NULL;
  void**  ppBuff = tSimpleHashGet(pSearchBuff, &groupId, sizeof(uint64_t));
  if (ppBuff) {
    pWinStates = (SArray*)(*ppBuff);
  } else {
    pWinStates = taosArrayInit(16, sizeof(SWinKey));
    QUERY_CHECK_NULL(pWinStates, code, lino, _end, terrno);

    code = tSimpleHashPut(pSearchBuff, &groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  (*ppResStates) = pWinStates;

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

static void setValueBuff(TSKEY ts, char* pVal, int32_t len, char* pBuff, int32_t buffLen) {
  SET_TSDATA_FLAG(pBuff, buffLen);
  if (len == 0) {
    *(TSKEY*)pBuff = ts;
    return;
  }
  memset(pBuff, 0, buffLen - 1);
  *(TSKEY*)pBuff = ts;
  memcpy(pBuff + sizeof(TSKEY), pVal, len);
}

int32_t getAndSetTsData(STableTsDataState* pTsDataState, uint64_t tableUid, TSKEY* pCurTs, void** ppCurPkVal,
                        TSKEY lastTs, void* pLastPkVal, int32_t lastPkLen, int32_t* pWinCode) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  bool    hasPk = (lastPkLen != 0);

  TSKEY* pDataVal = tSimpleHashGet(pTsDataState->pTableTsDataMap, &tableUid, sizeof(uint64_t));
  if (pDataVal != NULL) {
    (*pWinCode) = TSDB_CODE_SUCCESS;
    *pCurTs = *pDataVal;
    if ((*pCurTs) < lastTs) {
      setValueBuff(lastTs, pLastPkVal, lastPkLen, (char*)pDataVal, pTsDataState->pkValLen);
    } else {
      if (hasPk) {
        (*ppCurPkVal) = POINTER_SHIFT(pDataVal, sizeof(TSKEY));
        if ((*pCurTs) == lastTs && pTsDataState->comparePkColFn((*ppCurPkVal), pLastPkVal) < 0) {
          setValueBuff(lastTs, pLastPkVal, lastPkLen, (char*)pDataVal, pTsDataState->pkValLen);
        }
      }
    }
  } else {
    setValueBuff(lastTs, pLastPkVal, lastPkLen, pTsDataState->pPkValBuff, pTsDataState->pkValLen);
    int32_t size = tSimpleHashGetSize(pTsDataState->pTableTsDataMap);
    if (size < MAX_STATE_MAP_SIZE) {
      (*pWinCode) = TSDB_CODE_FAILED;
      code = tSimpleHashPut(pTsDataState->pTableTsDataMap, &tableUid, sizeof(uint64_t), pTsDataState->pPkValBuff,
                            pTsDataState->pkValLen);
      QUERY_CHECK_CODE(code, lino, _end);
    } else {
      (*pWinCode) = streamStateGetParTag_rocksdb(pTsDataState->pState, tableUid, &pTsDataState->pPkValBuff,
                                                 &pTsDataState->pkValLen);
      if ((*pWinCode) == TSDB_CODE_SUCCESS) {
        *pCurTs = *(TSKEY*)pTsDataState->pPkValBuff;
        if (hasPk) {
          (*ppCurPkVal) = POINTER_SHIFT(pTsDataState->pPkValBuff, sizeof(TSKEY));
        }
      }

      int32_t tmpCode = streamStatePutParTag_rocksdb(pTsDataState->pState, tableUid, pTsDataState->pPkValBuff,
                                                     pTsDataState->pkValLen);
    }
  }

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

int32_t doTsDataCommit(STableTsDataState* pTsDataState) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  void*   batch = NULL;
  char*   pTempBuf = NULL;

  batch = streamStateCreateBatch();
  QUERY_CHECK_NULL(batch, code, lino, _end, terrno);
  int           idx = streamStateGetCfIdx(pTsDataState->pState, "partag");
  int32_t       len = (pTsDataState->pkValLen + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;
  pTempBuf = taosMemoryCalloc(1, len);
  QUERY_CHECK_NULL(pTempBuf, code, lino, _end, terrno);

  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pTsDataState->pTableTsDataMap, pIte, &iter)) != NULL) {
    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {
      code = streamStatePutBatch_rocksdb(pTsDataState->pState, batch);
      streamStateClearBatch(batch);
      QUERY_CHECK_CODE(code, lino, _end);
    }

    if (HAS_TSDATA_FLAG(pIte, pTsDataState->pkValLen)) {
      void* pKey = tSimpleHashGetKey(pIte, NULL);
      UNSET_TSDATA_FLAG(pIte, pTsDataState->pkValLen);
      code = streamStatePutBatchOptimize(pTsDataState->pState, idx, batch, pKey, pIte, pTsDataState->pkValLen, 0,
                                         pTempBuf);
      QUERY_CHECK_CODE(code, lino, _end);
      memset(pTempBuf, 0, len);
      qDebug("flush ts data,table id:%" PRIu64 , *(uint64_t*)pKey);
    }
  }

  int32_t numOfElems = streamStateGetBatchSize(batch);
  if (numOfElems > 0) {
    code = streamStatePutBatch_rocksdb(pTsDataState->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));
  }
  taosMemoryFree(pTempBuf);
  streamStateDestroyBatch(batch);
  return code;
}

int32_t doRangeDataCommit(STableTsDataState* pTsDataState) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  void*   batch = NULL;

  batch = streamStateCreateBatch();
  QUERY_CHECK_NULL(batch, code, lino, _end, terrno);
  int           idx = streamStateGetCfIdx(pTsDataState->pState, "sess");
  int32_t       len = (pTsDataState->pkValLen + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;

  int32_t size = taosArrayGetSize(pTsDataState->pScanRanges);
  for (int32_t i = 0; i < size; i++) {
    SScanRange* pRange = taosArrayGet(pTsDataState->pScanRanges, i);
    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {
      code = streamStatePutBatch_rocksdb(pTsDataState->pState, batch);
      streamStateClearBatch(batch);
      QUERY_CHECK_CODE(code, lino, _end);
    }
    SSessionKey key = {.win = pRange->win, .groupId = 0};
    int32_t     uidSize = tSimpleHashGetSize(pRange->pUIds);
    int32_t     gpIdSize = tSimpleHashGetSize(pRange->pGroupIds);
    int32_t     size = uidSize + gpIdSize;
    uint64_t*   pIdBuf = (uint64_t*)taosMemoryCalloc(1, size);
    void*       pIte = NULL;
    int32_t     iter = 0;
    int32_t     i = 0;
    while ((pIte = tSimpleHashIterate(pTsDataState->pTableTsDataMap, pIte, &iter)) != NULL) {
      void* pTempKey = tSimpleHashGetKey(pIte, NULL);
      pIdBuf[i] = *(uint64_t*)pTempKey;
      i++;
    }

    code = streamStatePutBatchOptimize(pTsDataState->pState, idx, batch, &key, (void*)pIdBuf, size, 0,
                                       NULL);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  int32_t numOfElems = streamStateGetBatchSize(batch);
  if (numOfElems > 0) {
    code = streamStatePutBatch_rocksdb(pTsDataState->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);
  return code;
}

int32_t initTsDataState(STableTsDataState** ppTsDataState, int8_t pkType, int32_t pkLen, void* pState, void* pOtherState) {
  int32_t    code = TSDB_CODE_SUCCESS;
  int32_t    lino = 0;

  STableTsDataState* pTsDataState = taosMemoryCalloc(1, sizeof(STableTsDataState));
  QUERY_CHECK_NULL(pTsDataState, code, lino, _end, terrno);

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT);
  pTsDataState->pTableTsDataMap = tSimpleHashInit(DEFAULT_STATE_MAP_CAPACITY, hashFn);
  QUERY_CHECK_NULL(pTsDataState->pTableTsDataMap, code, lino, _end, terrno);

  pTsDataState->pkValLen = sizeof(TSKEY) + pkLen + sizeof(char);
  pTsDataState->pPkValBuff = taosMemoryCalloc(1, pTsDataState->pkValLen);
  QUERY_CHECK_NULL(pTsDataState->pPkValBuff, code, lino, _end, terrno);

  if (pkLen != 0) {
    pTsDataState->comparePkColFn = getKeyComparFunc(pkType, TSDB_ORDER_ASC);
  } else {
    pTsDataState->comparePkColFn = NULL;
  }

  pTsDataState->pScanRanges = taosArrayInit(64, sizeof(SScanRange));
  QUERY_CHECK_NULL(pTsDataState->pScanRanges, code, lino, _end, terrno);

  pTsDataState->pState = pState;
  pTsDataState->recValueLen = sizeof(SRecDataInfo) + pkLen;
  pTsDataState->pRecValueBuff = taosMemoryCalloc(1, pTsDataState->recValueLen);
  QUERY_CHECK_NULL(pTsDataState->pRecValueBuff, code, lino, _end, terrno);

  pTsDataState->curRecId = -1;

  pTsDataState->pStreamTaskState = pOtherState;

  pTsDataState->cfgIndex = streamStateGetCfIdx(pTsDataState->pState, "sess");
  pTsDataState->pBatch = streamStateCreateBatch();
  QUERY_CHECK_NULL(pTsDataState->pBatch, code, lino, _end, TSDB_CODE_FAILED);
  
  pTsDataState->batchBufflen = (pTsDataState->recValueLen + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;
  pTsDataState->pBatchBuff = taosMemoryCalloc(1, pTsDataState->batchBufflen);

  (*ppTsDataState) = pTsDataState;

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

static void destroyScanRange(SScanRange* pRange) {
  pRange->win.skey = INT64_MIN;
  pRange->win.ekey = INT64_MIN;
  tSimpleHashCleanup(pRange->pUIds);
  pRange->pUIds = NULL;
  tSimpleHashCleanup(pRange->pGroupIds);
  pRange->pGroupIds = NULL;
}

void destroyTsDataState(STableTsDataState* pTsDataState) {
  SArray* pScanRanges = pTsDataState->pScanRanges;
  int32_t size = taosArrayGetSize(pScanRanges);
  for (int32_t i = 0; i < size; i++) {
    SScanRange* pRange = taosArrayGet(pScanRanges, i);
    destroyScanRange(pRange);
  }
  taosArrayDestroy(pTsDataState->pScanRanges);
  tSimpleHashCleanup(pTsDataState->pTableTsDataMap);
  taosMemoryFreeClear(pTsDataState->pPkValBuff);
  taosMemoryFreeClear(pTsDataState->pState);
  taosMemoryFreeClear(pTsDataState->pRecValueBuff);
  pTsDataState->pStreamTaskState = NULL;

  streamStateClearBatch(pTsDataState->pBatch);
  streamStateDestroyBatch(pTsDataState->pBatch);
  pTsDataState->pBatch = NULL;
  taosMemoryFreeClear(pTsDataState->pBatchBuff);

  taosMemoryFreeClear(pTsDataState);
}

int32_t recoverTsData(STableTsDataState* pTsDataState) {
  int32_t          code = TSDB_CODE_SUCCESS;
  int32_t          lino = 0;
  SStreamStateCur* pCur = createStateCursor(NULL);
  streamStateParTagSeekKeyNext_rocksdb(pTsDataState->pState, INT64_MIN, pCur);
  while (1) {
    uint64_t tableUid = 0;
    void*    pVal = NULL;
    int32_t  len = 0;
    int32_t  winCode = streamStateParTagGetKVByCur_rocksdb(pCur, &tableUid, &pVal, &len);
    if (winCode != TSDB_CODE_SUCCESS) {
      break;
    }
    if (pTsDataState->pkValLen != len) {
      taosMemoryFree(pVal);
      streamStateCurNext_rocksdb(pCur);
      continue;
    }
    UNSET_TSDATA_FLAG(pVal, len);
    code = tSimpleHashPut(pTsDataState->pTableTsDataMap, &tableUid, sizeof(uint64_t), pVal, len);
    taosMemoryFree(pVal);
    QUERY_CHECK_CODE(code, lino, _end);
    streamStateCurNext_rocksdb(pCur);
  }

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

SStreamStateCur* getLastStateCur(SStreamFileState* pFileState, getStateBuffFn fn) {
  SStreamStateCur* pCur = createStateCursor(pFileState);
  if (pCur == NULL) {
    return NULL;
  }
  SSHashObj* pSearchBuff = fn(pFileState);
  pCur->buffIndex = 0;
  pCur->hashIter = 0;
  pCur->pHashData = NULL;
  pCur->pHashData = tSimpleHashIterate(pSearchBuff, pCur->pHashData, &pCur->hashIter);
  return pCur;
}

void moveLastStateCurNext(SStreamStateCur* pCur, getStateBuffFn fn) {
  SSHashObj* pSearchBuff = fn(pCur->pStreamFileState);
  pCur->pHashData = tSimpleHashIterate(pSearchBuff, pCur->pHashData, &pCur->hashIter);
}

int32_t getNLastStateKVByCur(SStreamStateCur* pCur, int32_t num, SArray* pRes) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  SArray*  pWinStates = NULL;
  int32_t size = 0;

  while(1) {
    if (pCur->pHashData == NULL) {
      return TSDB_CODE_FAILED;
    }
    pWinStates = *((void**)pCur->pHashData);
    size = taosArrayGetSize(pWinStates);
    if (size > 0) {
      break;
    }
    moveLastStateCurNext(pCur, getSearchBuff);
  }

  int32_t i = TMAX(size - num, 0);

  for ( ; i < size; i++) {
    SWinKey* pKey = taosArrayGet(pWinStates, i);
    int32_t  len = 0;
    void*    pVal = NULL;
    int32_t  winCode = TSDB_CODE_SUCCESS;
    code = addRowBuffIfNotExist(pCur->pStreamFileState, (void*)pKey, sizeof(SWinKey), &pVal, &len, &winCode);
    QUERY_CHECK_CODE(code, lino, _end);

    if (winCode != TSDB_CODE_SUCCESS) {
      qError("%s failed at line %d since window not exist. ts:%" PRId64 ",groupId:%" PRIu64, __func__, __LINE__,
             pKey->ts, pKey->groupId);
    }

    void* pTempRes = taosArrayPush(pRes, &pVal);
    QUERY_CHECK_NULL(pTempRes, code, lino, _end, terrno);
  }

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

int32_t reloadTsDataState(STableTsDataState* pTsDataState) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  STableTsDataState tmpState = *pTsDataState;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT);
  tmpState.pTableTsDataMap = tSimpleHashInit(DEFAULT_STATE_MAP_CAPACITY, hashFn);
  QUERY_CHECK_NULL(tmpState.pTableTsDataMap, code, lino, _end, terrno);

  code = recoverTsData(&tmpState);
  QUERY_CHECK_CODE(code, lino, _end);

  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pTsDataState->pTableTsDataMap, pIte, &iter)) != NULL) {
    size_t keyLen = 0;
    void* pKey = tSimpleHashGetKey(pIte, &keyLen);
    code = tSimpleHashPut(tmpState.pTableTsDataMap, pKey, keyLen, pIte, pTsDataState->pkValLen);
    QUERY_CHECK_CODE(code, lino, _end);
  }
  tSimpleHashCleanup(pTsDataState->pTableTsDataMap);
  pTsDataState->pTableTsDataMap = tmpState.pTableTsDataMap;

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

int32_t saveRecInfoToDisk(STableTsDataState* pTsDataState, SSessionKey* pKey, SRecDataInfo* pVal, int32_t vLen) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  SStateSessionKey stateKey = {.key = *pKey, .opNum = ((SStreamState*)pTsDataState->pState)->number};
  code = streamStatePutBatchOptimize(pTsDataState->pState, pTsDataState->cfgIndex, pTsDataState->pBatch, &stateKey, pVal, vLen, 0,
                                     pTsDataState->pBatchBuff);
  QUERY_CHECK_CODE(code, lino, _end);

  memset(pTsDataState->pBatchBuff, 0, pTsDataState->batchBufflen);

  if (streamStateGetBatchSize(pTsDataState->pBatch) >= BATCH_LIMIT) {
    code = streamStatePutBatch_rocksdb(pTsDataState->pState, pTsDataState->pBatch);
    streamStateClearBatch(pTsDataState->pBatch);
    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 flushRemainRecInfoToDisk(STableTsDataState* pTsDataState) {
  int32_t code = streamStatePutBatch_rocksdb(pTsDataState->pState, pTsDataState->pBatch);
  streamStateClearBatch(pTsDataState->pBatch);
  return code;
}

int32_t recoverHashSortBuff(SStreamFileState* pFileState, SArray* pWinStates, uint64_t groupId) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  SWinKey          start = {.groupId = groupId, .ts = INT64_MAX};
  void*            pState = getStateFileStore(pFileState);
  SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, &start);
  for (int32_t i = 0; i < NUM_OF_CACHE_WIN; i++) {
    SWinKey tmpKey = {.groupId = groupId};
    int32_t tmpRes = streamStateGetGroupKVByCur_rocksdb(pState, pCur, &tmpKey, NULL, 0);
    if (tmpRes != TSDB_CODE_SUCCESS) {
      break;
    }
    void* tmp = taosArrayPush(pWinStates, &tmpKey);
    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);
    streamStateCurPrev_rocksdb(pCur);
  }
  taosArraySort(pWinStates, winKeyCmprImpl);
  streamStateFreeCur(pCur);

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

int32_t getRowStateAllPrevRow(SStreamFileState* pFileState, const SWinKey* pKey, SArray* pResArray, int32_t maxNum) {
  int32_t    code = TSDB_CODE_SUCCESS;
  int32_t    lino = 0;
  SWinKey*   pPrevKey = NULL;
  SSHashObj* pSearchBuff = getSearchBuff(pFileState);
  void*      pState = getStateFileStore(pFileState);
  void**     ppBuff = (void**)tSimpleHashGet(pSearchBuff, &pKey->groupId, sizeof(uint64_t));
  int32_t    num = 0;
  if (ppBuff) {
    SArray* pWinStates = (SArray*)(*ppBuff);
    int32_t size = taosArrayGetSize(pWinStates);
    int32_t index = binarySearch(pWinStates, size, pKey, fillStateKeyCompare);
    for (; index >= 0 && num <= maxNum; index--) {
      pPrevKey = taosArrayGet(pWinStates, index);
      if (winKeyCmprImpl(pPrevKey, pKey) == 0) {
        continue;
      }
      void*   pVal = NULL;
      int32_t len = 0;
      int32_t winCode = TSDB_CODE_SUCCESS;
      code = addRowBuffIfNotExist(pFileState, (void*)pPrevKey, sizeof(SWinKey), &pVal, &len, &winCode);
      QUERY_CHECK_CODE(code, lino, _end);
      void* tempRes = taosArrayPush(pResArray, &pVal);
      QUERY_CHECK_NULL(tempRes, code, lino, _end, terrno);
      num++;
    }
  }

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

int32_t setStateRecFlag(SStreamFileState* pFileState, const void* pKey, int32_t keyLen, int32_t mode) {
  return tSimpleHashPut(pFileState->pRecFlagMap, pKey, keyLen, &mode, sizeof(int32_t));
}

int32_t getStateRecFlag(SStreamFileState* pFileState, const void* pKey, int32_t keyLen, int32_t* pMode) {
  void* pVal = tSimpleHashGet(pFileState->pRecFlagMap, pKey, keyLen);
  if (pVal == NULL) {
    return TSDB_CODE_FAILED;
  }
  *pMode = *(int32_t*) pVal;
  return TSDB_CODE_SUCCESS;
}

void clearExpiredSessionState(SStreamFileState* pFileState, int32_t numOfKeep, TSKEY minTs, SSHashObj* pFlushGroup) {
  int32_t    code = TSDB_CODE_SUCCESS;
  int32_t    lino = 0;
  SSHashObj* pSessionBuff = pFileState->rowStateBuff;
  SStreamSnapshot* pFlushList = NULL;
  if (pFlushGroup != NULL) {
    pFlushList = tdListNew(POINTER_BYTES);
  }
  void*      pIte = NULL;
  int32_t    iter = 0;
  while ((pIte = tSimpleHashIterate(pSessionBuff, pIte, &iter)) != NULL) {
    SArray* pWinStates = *((void**)pIte);
    int32_t arraySize = TARRAY_SIZE(pWinStates);
    if (minTs != INT64_MAX && arraySize > numOfKeep) {
      SSessionKey key = {.win.skey = minTs, .win.ekey = minTs};
      key.groupId = *(uint64_t*)tSimpleHashGetKey(pIte, NULL);
      int32_t index = binarySearch(pWinStates, arraySize, &key, fillStateKeyCompare);
      numOfKeep = TMAX(arraySize - index, MIN_NUM_OF_SORT_CACHE_WIN);
      qDebug("modify numOfKeep, numOfKeep:%d. %s at line %d", numOfKeep, __func__, __LINE__);
    }

    int32_t size = arraySize - numOfKeep;
    for (int32_t i = 0; i < size; i++) {
      SRowBuffPos* pPos = taosArrayGetP(pWinStates, i);
      SSessionKey* pKey = pPos->pKey;
      if (tSimpleHashGetSize(pFileState->pRecFlagMap) > 0) {
        tSimpleHashRemove(pFileState->pRecFlagMap, pKey, sizeof(SSessionKey));
      }
      pPos->invalid = true;

      if (i == 0 && pFlushGroup != NULL) {
        void* pGpVal = tSimpleHashGet(pFlushGroup, &pKey->groupId, sizeof(uint64_t));
        if (pGpVal == NULL) {
          code = tdListAppend(pFlushList, &pPos);
          QUERY_CHECK_CODE(code, lino, _end);
          code = tSimpleHashPut(pFlushGroup, &pKey->groupId, sizeof(uint64_t), NULL, 0);
          QUERY_CHECK_CODE(code, lino, _end);
          continue;
        }
      }
      pPos->beFlushed = true;
      qTrace("clear expired session buff, ts:%" PRId64 ",groupid:%" PRIu64 ". %s at line %d", pKey->win.skey, pKey->groupId, __func__, __LINE__);

      if (isFlushedState(pFileState, pKey->win.skey, 0)) {
        int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey);
        qTrace("clear expired file, ts:%" PRId64 ". %s at line %d res:%d", pKey->win.skey, __func__, __LINE__, code_file);
      }
    }
    taosArrayRemoveBatch(pWinStates, 0, size, NULL);
  }

  if (pFlushList != NULL) {
    flushSnapshot(pFileState, pFlushList, false);
    code = clearRowBuffNonFlush(pFileState);
    QUERY_CHECK_CODE(code, lino, _end);
    tdListFreeP(pFlushList, destroyRowBuffPosPtr);
  }

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

taosdata / TDengine / #3660

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