tstreamFileState.c 15.7 KB
Newer Older
5
54liuyao 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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"

dengyihao's avatar
dengyihao 已提交
18
#include "query.h"
dengyihao's avatar
dengyihao 已提交
19
#include "streamBackendRocksdb.h"
5
54liuyao 已提交
20
#include "taos.h"
dengyihao's avatar
dengyihao 已提交
21
#include "tcommon.h"
5
54liuyao 已提交
22 23
#include "thash.h"
#include "tsimplehash.h"
24
#include "storageapi.h"
5
54liuyao 已提交
25

L
add ini  
liuyao 已提交
26
#define FLUSH_RATIO                    0.5
dengyihao's avatar
dengyihao 已提交
27
#define FLUSH_NUM                      4
5
54liuyao 已提交
28 29 30 31 32 33 34 35
#define DEFAULT_MAX_STREAM_BUFFER_SIZE (128 * 1024 * 1024);

struct SStreamFileState {
  SList*     usedBuffs;
  SList*     freeBuffs;
  SSHashObj* rowBuffMap;
  void*      pFileStore;
  int32_t    rowSize;
L
liuyao 已提交
36
  int32_t    selectivityRowSize;
5
54liuyao 已提交
37 38 39 40 41
  int32_t    keyLen;
  uint64_t   preCheckPointVersion;
  uint64_t   checkPointVersion;
  TSKEY      maxTs;
  TSKEY      deleteMark;
5
54liuyao 已提交
42
  TSKEY      flushMark;
5
54liuyao 已提交
43 44
  uint64_t   maxRowCount;
  uint64_t   curRowCount;
5
54liuyao 已提交
45
  GetTsFun   getTs;
5
54liuyao 已提交
46 47 48 49
};

typedef SRowBuffPos SRowBuffInfo;

dengyihao's avatar
dengyihao 已提交
50 51
SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize,
                                      GetTsFun fp, void* pFile, TSKEY delMark) {
5
54liuyao 已提交
52 53 54 55 56 57 58 59 60 61 62
  if (memSize <= 0) {
    memSize = DEFAULT_MAX_STREAM_BUFFER_SIZE;
  }
  if (rowSize == 0) {
    goto _error;
  }

  SStreamFileState* pFileState = taosMemoryCalloc(1, sizeof(SStreamFileState));
  if (!pFileState) {
    goto _error;
  }
L
liuyao 已提交
63
  rowSize += selectRowSize;
dengyihao's avatar
dengyihao 已提交
64
  pFileState->maxRowCount = TMAX((uint64_t)memSize / rowSize, FLUSH_NUM * 2);
5
54liuyao 已提交
65 66 67
  pFileState->usedBuffs = tdListNew(POINTER_BYTES);
  pFileState->freeBuffs = tdListNew(POINTER_BYTES);
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
dengyihao's avatar
dengyihao 已提交
68
  int32_t    cap = TMIN(10240, pFileState->maxRowCount);
L
liuyao 已提交
69
  pFileState->rowBuffMap = tSimpleHashInit(cap, hashFn);
5
54liuyao 已提交
70 71 72
  if (!pFileState->usedBuffs || !pFileState->freeBuffs || !pFileState->rowBuffMap) {
    goto _error;
  }
L
liuyao 已提交
73
  pFileState->keyLen = keySize;
5
54liuyao 已提交
74
  pFileState->rowSize = rowSize;
L
liuyao 已提交
75
  pFileState->selectivityRowSize = selectRowSize;
5
54liuyao 已提交
76 77 78
  pFileState->preCheckPointVersion = 0;
  pFileState->checkPointVersion = 1;
  pFileState->pFileStore = pFile;
5
54liuyao 已提交
79
  pFileState->getTs = fp;
5
54liuyao 已提交
80
  pFileState->curRowCount = 0;
5
54liuyao 已提交
81
  pFileState->deleteMark = delMark;
L
add ini  
liuyao 已提交
82 83
  pFileState->flushMark = INT64_MIN;
  pFileState->maxTs = INT64_MIN;
L
liuyao 已提交
84
  recoverSnapshot(pFileState);
5
54liuyao 已提交
85 86 87
  return pFileState;

_error:
5
54liuyao 已提交
88
  streamFileStateDestroy(pFileState);
5
54liuyao 已提交
89 90 91 92
  return NULL;
}

void destroyRowBuffPos(SRowBuffPos* pPos) {
5
54liuyao 已提交
93
  taosMemoryFreeClear(pPos->pKey);
5
54liuyao 已提交
94 95 96 97 98 99 100 101
  taosMemoryFreeClear(pPos->pRowBuff);
  taosMemoryFree(pPos);
}

void destroyRowBuffPosPtr(void* ptr) {
  if (!ptr) {
    return;
  }
5
54liuyao 已提交
102
  SRowBuffPos* pPos = *(SRowBuffPos**)ptr;
L
liuyao 已提交
103 104 105
  if (!pPos->beUsed) {
    destroyRowBuffPos(pPos);
  }
5
54liuyao 已提交
106 107
}

L
liuyao 已提交
108 109 110 111 112 113 114 115
void destroyRowBuffAllPosPtr(void* ptr) {
  if (!ptr) {
    return;
  }
  SRowBuffPos* pPos = *(SRowBuffPos**)ptr;
  destroyRowBuffPos(pPos);
}

5
54liuyao 已提交
116 117 118 119 120 121 122 123 124 125 126
void destroyRowBuff(void* ptr) {
  if (!ptr) {
    return;
  }
  taosMemoryFree(*(void**)ptr);
}

void streamFileStateDestroy(SStreamFileState* pFileState) {
  if (!pFileState) {
    return;
  }
L
liuyao 已提交
127
  tdListFreeP(pFileState->usedBuffs, destroyRowBuffAllPosPtr);
5
54liuyao 已提交
128
  tdListFreeP(pFileState->freeBuffs, destroyRowBuff);
5
54liuyao 已提交
129
  tSimpleHashCleanup(pFileState->rowBuffMap);
5
54liuyao 已提交
130
  taosMemoryFree(pFileState);
5
54liuyao 已提交
131 132
}

5
54liuyao 已提交
133
void clearExpiredRowBuff(SStreamFileState* pFileState, TSKEY ts, bool all) {
5
54liuyao 已提交
134 135 136 137 138
  SListIter iter = {0};
  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&iter)) != NULL) {
5
54liuyao 已提交
139
    SRowBuffPos* pPos = *(SRowBuffPos**)(pNode->data);
dengyihao's avatar
dengyihao 已提交
140
    if (all || (pFileState->getTs(pPos->pKey) < ts)) {
L
liuyao 已提交
141
      ASSERT(pPos->pRowBuff != NULL);
5
54liuyao 已提交
142
      tdListAppend(pFileState->freeBuffs, &(pPos->pRowBuff));
5
54liuyao 已提交
143
      pPos->pRowBuff = NULL;
L
liuyao 已提交
144 145 146
      if (!all) {
        tSimpleHashRemove(pFileState->rowBuffMap, pPos->pKey, pFileState->keyLen);
      }
5
54liuyao 已提交
147
      destroyRowBuffPos(pPos);
L
liuyao 已提交
148 149
      tdListPopNode(pFileState->usedBuffs, pNode);
      taosMemoryFreeClear(pNode);
5
54liuyao 已提交
150 151 152 153
    }
  }
}

5
54liuyao 已提交
154
void streamFileStateClear(SStreamFileState* pFileState) {
L
add ini  
liuyao 已提交
155 156
  pFileState->flushMark = INT64_MIN;
  pFileState->maxTs = INT64_MIN;
5
54liuyao 已提交
157 158 159 160
  tSimpleHashClear(pFileState->rowBuffMap);
  clearExpiredRowBuff(pFileState, 0, true);
}

dengyihao's avatar
dengyihao 已提交
161
bool needClearDiskBuff(SStreamFileState* pFileState) { return pFileState->flushMark > 0; }
L
liuyao 已提交
162

L
liuyao 已提交
163
void popUsedBuffs(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uint64_t max, bool used) {
dengyihao's avatar
dengyihao 已提交
164
  uint64_t  i = 0;
5
54liuyao 已提交
165 166 167 168
  SListIter iter = {0};
  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

  SListNode* pNode = NULL;
L
liuyao 已提交
169
  while ((pNode = tdListNext(&iter)) != NULL && i < max) {
5
54liuyao 已提交
170
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
L
liuyao 已提交
171
    if (pPos->beUsed == used) {
5
54liuyao 已提交
172
      tdListAppend(pFlushList, &pPos);
5
54liuyao 已提交
173 174
      pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));
      tSimpleHashRemove(pFileState->rowBuffMap, pPos->pKey, pFileState->keyLen);
L
liuyao 已提交
175 176
      tdListPopNode(pFileState->usedBuffs, pNode);
      taosMemoryFreeClear(pNode);
5
54liuyao 已提交
177 178 179
      i++;
    }
  }
L
liuyao 已提交
180
  qInfo("do stream state flush %d rows to disck. is used: %d", listNEles(pFlushList), used);
L
liuyao 已提交
181 182 183 184 185 186 187 188 189 190 191 192 193
}

int32_t flushRowBuff(SStreamFileState* pFileState) {
  SStreamSnapshot* pFlushList = tdListNew(POINTER_BYTES);
  if (!pFlushList) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  uint64_t num = (uint64_t)(pFileState->curRowCount * FLUSH_RATIO);
  num = TMAX(num, FLUSH_NUM);
  popUsedBuffs(pFileState, pFlushList, num, false);
  if (isListEmpty(pFlushList)) {
    popUsedBuffs(pFileState, pFlushList, num, true);
  }
L
liuyao 已提交
194
  flushSnapshot(pFileState, pFlushList, false);
L
liuyao 已提交
195 196 197 198 199 200 201 202 203 204
  SListIter fIter = {0};
  tdListInitIter(pFlushList, &fIter, TD_LIST_FORWARD);
  SListNode* pNode = NULL;
  while ((pNode = tdListNext(&fIter)) != NULL) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
    ASSERT(pPos->pRowBuff != NULL);
    tdListAppend(pFileState->freeBuffs, &pPos->pRowBuff);
    pPos->pRowBuff = NULL;
  }
  tdListFreeP(pFlushList, destroyRowBuffPosPtr);
5
54liuyao 已提交
205 206 207 208
  return TSDB_CODE_SUCCESS;
}

int32_t clearRowBuff(SStreamFileState* pFileState) {
5
54liuyao 已提交
209
  clearExpiredRowBuff(pFileState, pFileState->maxTs - pFileState->deleteMark, false);
5
54liuyao 已提交
210 211 212 213 214 215
  if (isListEmpty(pFileState->freeBuffs)) {
    return flushRowBuff(pFileState);
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
216
void* getFreeBuff(SList* lists, int32_t buffSize) {
5
54liuyao 已提交
217 218 219 220 221
  SListNode* pNode = tdListPopHead(lists);
  if (!pNode) {
    return NULL;
  }
  void* ptr = *(void**)pNode->data;
5
54liuyao 已提交
222
  memset(ptr, 0, buffSize);
5
54liuyao 已提交
223 224 225 226 227 228
  taosMemoryFree(pNode);
  return ptr;
}

SRowBuffPos* getNewRowPos(SStreamFileState* pFileState) {
  SRowBuffPos* pPos = taosMemoryCalloc(1, sizeof(SRowBuffPos));
L
liuyao 已提交
229
  pPos->pKey = taosMemoryCalloc(1, pFileState->keyLen);
5
54liuyao 已提交
230
  void* pBuff = getFreeBuff(pFileState->freeBuffs, pFileState->rowSize);
5
54liuyao 已提交
231 232
  if (pBuff) {
    pPos->pRowBuff = pBuff;
L
liuyao 已提交
233
    goto _end;
5
54liuyao 已提交
234 235 236 237 238 239 240
  }

  if (pFileState->curRowCount < pFileState->maxRowCount) {
    pBuff = taosMemoryCalloc(1, pFileState->rowSize);
    if (pBuff) {
      pPos->pRowBuff = pBuff;
      pFileState->curRowCount++;
L
liuyao 已提交
241
      goto _end;
5
54liuyao 已提交
242 243 244 245 246
    }
  }

  int32_t code = clearRowBuff(pFileState);
  ASSERT(code == 0);
5
54liuyao 已提交
247
  pPos->pRowBuff = getFreeBuff(pFileState->freeBuffs, pFileState->rowSize);
L
liuyao 已提交
248 249 250 251

_end:
  tdListAppend(pFileState->usedBuffs, &pPos);
  ASSERT(pPos->pRowBuff != NULL);
5
54liuyao 已提交
252 253 254 255
  return pPos;
}

int32_t getRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen) {
5
54liuyao 已提交
256
  pFileState->maxTs = TMAX(pFileState->maxTs, pFileState->getTs(pKey));
5
54liuyao 已提交
257 258
  SRowBuffPos** pos = tSimpleHashGet(pFileState->rowBuffMap, pKey, keyLen);
  if (pos) {
L
liuyao 已提交
259 260 261
    *pVLen = pFileState->rowSize;
    *pVal = *pos;
    (*pos)->beUsed = true;
5
54liuyao 已提交
262 263 264
    return TSDB_CODE_SUCCESS;
  }
  SRowBuffPos* pNewPos = getNewRowPos(pFileState);
L
liuyao 已提交
265 266
  pNewPos->beUsed = true;
  ASSERT(pNewPos->pRowBuff);
5
54liuyao 已提交
267 268 269 270 271
  memcpy(pNewPos->pKey, pKey, keyLen);

  TSKEY ts = pFileState->getTs(pKey);
  if (ts > pFileState->maxTs - pFileState->deleteMark && ts < pFileState->flushMark) {
    int32_t len = 0;
dengyihao's avatar
dengyihao 已提交
272
    void*   pVal = NULL;
L
liuyao 已提交
273
    int32_t code = streamStateGet_rocksdb(pFileState->pFileStore, pKey, &pVal, &len);
L
add ini  
liuyao 已提交
274
    qDebug("===stream===get %" PRId64 " from disc, res %d", ts, code);
L
liuyao 已提交
275 276 277
    if (code == TSDB_CODE_SUCCESS) {
      memcpy(pNewPos->pRowBuff, pVal, len);
    }
5
54liuyao 已提交
278 279 280
    taosMemoryFree(pVal);
  }

5
54liuyao 已提交
281
  tSimpleHashPut(pFileState->rowBuffMap, pKey, keyLen, &pNewPos, POINTER_BYTES);
5
54liuyao 已提交
282 283 284 285
  if (pVal) {
    *pVLen = pFileState->rowSize;
    *pVal = pNewPos;
  }
5
54liuyao 已提交
286 287 288
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
289 290 291 292 293 294 295
int32_t deleteRowBuff(SStreamFileState* pFileState, const void* pKey, int32_t keyLen) {
  int32_t code_buff = tSimpleHashRemove(pFileState->rowBuffMap, pKey, keyLen);
  int32_t code_rocks = streamStateDel_rocksdb(pFileState->pFileStore, pKey);
  return code_buff == TSDB_CODE_SUCCESS ? code_buff : code_rocks;
}

int32_t getRowBuffByPos(SStreamFileState* pFileState, SRowBuffPos* pPos, void** pVal) {
5
54liuyao 已提交
296
  if (pPos->pRowBuff) {
5
54liuyao 已提交
297 298
    (*pVal) = pPos->pRowBuff;
    return TSDB_CODE_SUCCESS;
5
54liuyao 已提交
299 300
  }

5
54liuyao 已提交
301
  pPos->pRowBuff = getFreeBuff(pFileState->freeBuffs, pFileState->rowSize);
L
liuyao 已提交
302 303 304 305 306 307
  if (!pPos->pRowBuff) {
    int32_t code = clearRowBuff(pFileState);
    ASSERT(code == 0);
    pPos->pRowBuff = getFreeBuff(pFileState->freeBuffs, pFileState->rowSize);
    ASSERT(pPos->pRowBuff);
  }
5
54liuyao 已提交
308 309

  int32_t len = 0;
dengyihao's avatar
dengyihao 已提交
310
  void*   pBuff = NULL;
L
liuyao 已提交
311 312 313
  streamStateGet_rocksdb(pFileState->pFileStore, pPos->pKey, &pBuff, &len);
  memcpy(pPos->pRowBuff, pBuff, len);
  taosMemoryFree(pBuff);
5
54liuyao 已提交
314
  (*pVal) = pPos->pRowBuff;
L
liuyao 已提交
315
  tdListPrepend(pFileState->usedBuffs, &pPos);
5
54liuyao 已提交
316 317 318 319 320 321 322 323 324
  return TSDB_CODE_SUCCESS;
}

bool hasRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen) {
  SRowBuffPos** pos = tSimpleHashGet(pFileState->rowBuffMap, pKey, keyLen);
  if (pos) {
    return true;
  }
  return false;
5
54liuyao 已提交
325 326
}

dengyihao's avatar
dengyihao 已提交
327
void releaseRowBuffPos(SRowBuffPos* pBuff) { pBuff->beUsed = false; }
5
54liuyao 已提交
328 329

SStreamSnapshot* getSnapshot(SStreamFileState* pFileState) {
dengyihao's avatar
dengyihao 已提交
330 331 332
  int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs) ? INT64_MIN
                                                                           : pFileState->maxTs - pFileState->deleteMark;
  clearExpiredRowBuff(pFileState, mark, false);
5
54liuyao 已提交
333 334 335
  return pFileState->usedBuffs;
}

dengyihao's avatar
dengyihao 已提交
336 337 338
void streamFileStateDecode(TSKEY* key, void* pBuff, int32_t len) { pBuff = taosDecodeFixedI64(pBuff, key); }

void streamFileStateEncode(TSKEY* key, void** pVal, int32_t* pLen) {
L
liuyao 已提交
339
  *pLen = sizeof(TSKEY);
L
liuyao 已提交
340
  (*pVal) = taosMemoryCalloc(1, *pLen);
dengyihao's avatar
dengyihao 已提交
341
  void* buff = *pVal;
dengyihao's avatar
dengyihao 已提交
342
  taosEncodeFixedI64(&buff, *key);
L
liuyao 已提交
343 344 345
}

int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState) {
dengyihao's avatar
dengyihao 已提交
346
  int32_t   code = TSDB_CODE_SUCCESS;
5
54liuyao 已提交
347 348 349
  SListIter iter = {0};
  tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD);

dengyihao's avatar
dengyihao 已提交
350
  const int32_t BATCH_LIMIT = 256;
dengyihao's avatar
dengyihao 已提交
351 352 353
  SListNode*    pNode = NULL;

  void* batch = streamStateCreateBatch();
5
54liuyao 已提交
354 355
  while ((pNode = tdListNext(&iter)) != NULL && code == TSDB_CODE_SUCCESS) {
    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;
L
liuyao 已提交
356
    ASSERT(pPos->pRowBuff && pFileState->rowSize > 0);
dengyihao's avatar
dengyihao 已提交
357 358 359 360
    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {
      code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
      streamStateClearBatch(batch);
    }
dengyihao's avatar
dengyihao 已提交
361 362

    SStateKey sKey = {.key = *((SWinKey*)pPos->pKey), .opNum = ((SStreamState*)pFileState->pFileStore)->number};
dengyihao's avatar
dengyihao 已提交
363
    code = streamStatePutBatch(pFileState->pFileStore, "state", batch, &sKey, pPos->pRowBuff, pFileState->rowSize, 0);
L
add ini  
liuyao 已提交
364
    qDebug("===stream===put %" PRId64 " to disc, res %d", sKey.key.ts, code);
L
liuyao 已提交
365
  }
dengyihao's avatar
dengyihao 已提交
366 367 368
  if (streamStateGetBatchSize(batch) > 0) {
    code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
  }
dengyihao's avatar
dengyihao 已提交
369
  streamStateClearBatch(batch);
dengyihao's avatar
dengyihao 已提交
370

L
liuyao 已提交
371
  if (flushState) {
dengyihao's avatar
dengyihao 已提交
372 373 374 375 376 377 378
    const char* taskKey = "streamFileState";
    {
      char    keyBuf[128] = {0};
      void*   valBuf = NULL;
      int32_t len = 0;
      sprintf(keyBuf, "%s:%" PRId64 "", taskKey, ((SStreamState*)pFileState->pFileStore)->checkPointId);
      streamFileStateEncode(&pFileState->flushMark, &valBuf, &len);
dengyihao's avatar
dengyihao 已提交
379
      code = streamStatePutBatch(pFileState->pFileStore, "default", batch, keyBuf, valBuf, len, 0);
dengyihao's avatar
dengyihao 已提交
380 381 382 383 384 385
      taosMemoryFree(valBuf);
    }
    {
      char    keyBuf[128] = {0};
      char    valBuf[64] = {0};
      int32_t len = 0;
dengyihao's avatar
dengyihao 已提交
386
      memcpy(keyBuf, taskKey, strlen(taskKey));
dengyihao's avatar
dengyihao 已提交
387
      len = sprintf(valBuf, "%" PRId64 "", ((SStreamState*)pFileState->pFileStore)->checkPointId);
dengyihao's avatar
dengyihao 已提交
388
      code = streamStatePutBatch(pFileState->pFileStore, "default", batch, keyBuf, valBuf, len, 0);
dengyihao's avatar
dengyihao 已提交
389 390
    }
    streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);
5
54liuyao 已提交
391
  }
dengyihao's avatar
dengyihao 已提交
392
  streamStateDestroyBatch(batch);
dengyihao's avatar
dengyihao 已提交
393

5
54liuyao 已提交
394 395
  return code;
}
L
liuyao 已提交
396

dengyihao's avatar
dengyihao 已提交
397 398 399 400 401 402
int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {
  const char* taskKey = "streamFileState";
  char        keyBuf[128] = {0};
  sprintf(keyBuf, "%s:%" PRId64 "", taskKey, checkpointId);
  return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);
}
L
liuyao 已提交
403

dengyihao's avatar
dengyihao 已提交
404 405
int32_t getSnapshotIdList(SStreamFileState* pFileState, SArray* list) {
  const char* taskKey = "streamFileState";
dengyihao's avatar
dengyihao 已提交
406
  return streamDefaultIterGet_rocksdb(pFileState->pFileStore, taskKey, NULL, list);
dengyihao's avatar
dengyihao 已提交
407
}
L
liuyao 已提交
408 409

int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
dengyihao's avatar
dengyihao 已提交
410 411 412 413 414 415 416
  int32_t     code = TSDB_CODE_SUCCESS;
  const char* taskKey = "streamFileState";
  int64_t     maxCheckPointId = 0;
  {
    char    buf[128] = {0};
    void*   val = NULL;
    int32_t len = 0;
dengyihao's avatar
dengyihao 已提交
417
    memcpy(buf, taskKey, strlen(taskKey));
dengyihao's avatar
dengyihao 已提交
418 419 420 421 422 423
    code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
    if (code != 0) {
      return TSDB_CODE_FAILED;
    }
    sscanf(val, "%" PRId64 "", &maxCheckPointId);
  }
dengyihao's avatar
dengyihao 已提交
424
  for (int64_t i = maxCheckPointId; i > 0; i--) {
dengyihao's avatar
dengyihao 已提交
425 426 427 428 429 430 431 432 433 434
    char    buf[128] = {0};
    void*   val = 0;
    int32_t len = 0;
    sprintf(buf, "%s:%" PRId64 "", taskKey, i);
    code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);
    if (code != 0) {
      return TSDB_CODE_FAILED;
    }
    TSKEY ts;
    sscanf(val, "%" PRId64 "", &ts);
L
liuyao 已提交
435
    if (ts < mark) {
dengyihao's avatar
dengyihao 已提交
436
      // statekey winkey.ts < mark
dengyihao's avatar
dengyihao 已提交
437
      forceRemoveCheckpoint(pFileState, i);
dengyihao's avatar
dengyihao 已提交
438
      break;
dengyihao's avatar
dengyihao 已提交
439
    } else {
dengyihao's avatar
dengyihao 已提交
440 441
    }
  }
L
liuyao 已提交
442 443
  return code;
}
dengyihao's avatar
dengyihao 已提交
444

L
liuyao 已提交
445 446
int32_t recoverSnapshot(SStreamFileState* pFileState) {
  int32_t code = TSDB_CODE_SUCCESS;
L
liuyao 已提交
447
  if (pFileState->maxTs != INT64_MIN) {
dengyihao's avatar
dengyihao 已提交
448 449 450 451
    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)
                       ? INT64_MIN
                       : pFileState->maxTs - pFileState->deleteMark;
    deleteExpiredCheckPoint(pFileState, mark);
L
liuyao 已提交
452
  }
dengyihao's avatar
dengyihao 已提交
453
  void*   pStVal = NULL;
L
liuyao 已提交
454 455
  int32_t len = 0;

dengyihao's avatar
dengyihao 已提交
456 457
  SWinKey          key = {.groupId = 0, .ts = 0};
  SStreamStateCur* pCur = streamStateSeekToLast_rocksdb(pFileState->pFileStore, &key);
dengyihao's avatar
dengyihao 已提交
458 459
  if (pCur == NULL) {
    return -1;
L
liuyao 已提交
460
  }
dengyihao's avatar
dengyihao 已提交
461

L
liuyao 已提交
462 463 464 465
  while (code == TSDB_CODE_SUCCESS) {
    if (pFileState->curRowCount == pFileState->maxRowCount) {
      break;
    }
dengyihao's avatar
dengyihao 已提交
466 467
    void*        pVal = NULL;
    int32_t      pVLen = 0;
L
liuyao 已提交
468
    SRowBuffPos* pNewPos = getNewRowPos(pFileState);
dengyihao's avatar
dengyihao 已提交
469
    code = streamStateGetKVByCur_rocksdb(pCur, pNewPos->pKey, (const void**)&pVal, &pVLen);
L
liuyao 已提交
470 471
    if (code != TSDB_CODE_SUCCESS || pFileState->getTs(pNewPos->pKey) < pFileState->flushMark) {
      destroyRowBuffPos(pNewPos);
L
liuyao 已提交
472 473
      SListNode* pNode = tdListPopTail(pFileState->usedBuffs);
      taosMemoryFreeClear(pNode);
L
liuyao 已提交
474 475 476 477 478 479 480 481 482 483
      break;
    }
    memcpy(pNewPos->pRowBuff, pVal, pVLen);
    code = tSimpleHashPut(pFileState->rowBuffMap, pNewPos->pKey, pFileState->rowSize, &pNewPos, POINTER_BYTES);
    if (code != TSDB_CODE_SUCCESS) {
      destroyRowBuffPos(pNewPos);
      break;
    }
    code = streamStateCurPrev_rocksdb(pFileState->pFileStore, pCur);
  }
dengyihao's avatar
dengyihao 已提交
484
  streamStateFreeCur(pCur);
L
liuyao 已提交
485

5
54liuyao 已提交
486
  return TSDB_CODE_SUCCESS;
L
liuyao 已提交
487 488
}

dengyihao's avatar
dengyihao 已提交
489
int32_t streamFileStateGeSelectRowSize(SStreamFileState* pFileState) { return pFileState->selectivityRowSize; }