taosdata/TDengine | Build 3621 | source/libs/stream/src/tstreamFileState.c | Coveralls

UNCOV

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

×

UNCOV

75

  SWinKey* pWin2 = taosArrayGet(pDatas, pos);

×

UNCOV

76

  return winKeyCmprImpl((SWinKey*)pWin1, pWin2);

×

UNCOV

79

int32_t stateHashBuffRemoveFn(void* pBuff, const void* pKey, size_t keyLen) {

×

UNCOV

80

  SRowBuffPos** pos = tSimpleHashGet(pBuff, pKey, keyLen);

×

UNCOV

81

  if (pos) {

×

UNCOV

82

    (*pos)->beFlushed = true;

×

UNCOV

84

  return tSimpleHashRemove(pBuff, pKey, keyLen);

×

UNCOV

87

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

×

UNCOV

88

  size_t        keyLen = pFileState->keyLen;

×

UNCOV

89

  SRowBuffPos** ppPos = tSimpleHashGet(pFileState->rowStateBuff, pPos->pKey, keyLen);

×

UNCOV

90

  if (ppPos) {

×

UNCOV

91

    if ((*ppPos) == pPos) {

×

UNCOV

92

      int32_t tmpRes = tSimpleHashRemove(pFileState->rowStateBuff, pPos->pKey, keyLen);

×

UNCOV

93

      qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);

×

UNCOV

96

×

98

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

×

UNCOV

100

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

×

UNCOV

102

int32_t intervalFileRemoveFn(SStreamFileState* pFileState, const void* pKey) {

×

UNCOV

103

  return streamStateDel_rocksdb(pFileState->pFileStore, pKey);

×

UNCOV

106

int32_t intervalFileGetFn(SStreamFileState* pFileState, void* pKey, void** data, int32_t* pDataLen) {

×

UNCOV

107

  return streamStateGet_rocksdb(pFileState->pFileStore, pKey, data, pDataLen);

×

UNCOV

110

void* intervalCreateStateKey(SRowBuffPos* pPos, int64_t num) {

×

UNCOV

111

  SStateKey* pStateKey = taosMemoryCalloc(1, sizeof(SStateKey));

×

UNCOV

112

  if (pStateKey == NULL) {

×

113

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

×

114

    return NULL;

×

UNCOV

116

  SWinKey* pWinKey = pPos->pKey;

×

UNCOV

117

  pStateKey->key = *pWinKey;

×

UNCOV

118

  pStateKey->opNum = num;

×

UNCOV

119

  return pStateKey;

×

UNCOV

122

void* defaultCreateStateKey(SRowBuffPos* pPos, int64_t num) {

×

UNCOV

123

  SWinKey* pStateKey = taosMemoryCalloc(1, sizeof(SWinKey));

×

UNCOV

124

  if (pStateKey == NULL) {

×

125

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

×

126

    return NULL;

×

UNCOV

128

  SWinKey* pWinKey = pPos->pKey;

×

UNCOV

129

  *pStateKey = *pWinKey;

×

UNCOV

130

  return pStateKey;

×

UNCOV

133

int32_t sessionFileRemoveFn(SStreamFileState* pFileState, const void* pKey) {

×

UNCOV

134

  return streamStateSessionDel_rocksdb(pFileState->pFileStore, pKey);

×

UNCOV

137

int32_t sessionFileGetFn(SStreamFileState* pFileState, void* pKey, void** data, int32_t* pDataLen) {

×

UNCOV

138

  return streamStateSessionGet_rocksdb(pFileState->pFileStore, pKey, data, pDataLen);

×

UNCOV

141

void* sessionCreateStateKey(SRowBuffPos* pPos, int64_t num) {

×

UNCOV

142

  SStateSessionKey* pStateKey = taosMemoryCalloc(1, sizeof(SStateSessionKey));

×

UNCOV

143

  if (pStateKey == NULL) {

×

144

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

×

145

    return NULL;

×

UNCOV

147

  SSessionKey* pWinKey = pPos->pKey;

×

UNCOV

148

  pStateKey->key = *pWinKey;

×

UNCOV

149

  pStateKey->opNum = num;

×

UNCOV

150

  return pStateKey;

×

UNCOV

153

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

×

UNCOV

155

static int32_t streamFileStateEncode(TSKEY* pKey, void** pVal, int32_t* pLen) {

×

UNCOV

156

  *pLen = sizeof(TSKEY);

×

UNCOV

157

  (*pVal) = taosMemoryCalloc(1, *pLen);

×

UNCOV

158

  if ((*pVal) == NULL) {

×

159

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

×

160

    return terrno;

×

UNCOV

162

  void*   buff = *pVal;

×

UNCOV

163

  int32_t tmp = taosEncodeFixedI64(&buff, *pKey);

×

UNCOV

164

  return TSDB_CODE_SUCCESS;

×

UNCOV

167

int32_t streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_t rowSize, uint32_t selectRowSize, GetTsFun fp,

×

UNCOV

170

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

171

  int32_t lino = 0;

×

UNCOV

172

  if (memSize <= 0) {

×

173

    memSize = DEFAULT_MAX_STREAM_BUFFER_SIZE;

×

UNCOV

175

  if (rowSize == 0) {

×

176

    code = TSDB_CODE_INVALID_PARA;

×

177

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

180

  SStreamFileState* pFileState = taosMemoryCalloc(1, sizeof(SStreamFileState));

×

UNCOV

181

  QUERY_CHECK_NULL(pFileState, code, lino, _end, terrno);

×

UNCOV

183

  rowSize += selectRowSize;

×

UNCOV

184

  pFileState->maxRowCount = TMAX((uint64_t)memSize / rowSize, FLUSH_NUM * 2);

×

UNCOV

185

  pFileState->usedBuffs = tdListNew(POINTER_BYTES);

×

UNCOV

186

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

×

UNCOV

188

  pFileState->freeBuffs = tdListNew(POINTER_BYTES);

×

UNCOV

189

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

×

UNCOV

191

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

UNCOV

192

  int32_t    cap = TMIN(MIN_NUM_OF_ROW_BUFF, pFileState->maxRowCount);

×

UNCOV

193

  if (type == STREAM_STATE_BUFF_HASH || type == STREAM_STATE_BUFF_HASH_SEARCH) {

×

UNCOV

194

    pFileState->rowStateBuff = tSimpleHashInit(cap, hashFn);

×

UNCOV

195

    pFileState->stateBuffCleanupFn = stateHashBuffCleanupFn;

×

UNCOV

196

    pFileState->stateBuffRemoveFn = stateHashBuffRemoveFn;

×

UNCOV

197

    pFileState->stateBuffRemoveByPosFn = stateHashBuffRemoveByPosFn;

×

UNCOV

198

    pFileState->stateBuffCreateStateKeyFn = intervalCreateStateKey;

×

UNCOV

200

    pFileState->stateFileRemoveFn = intervalFileRemoveFn;

×

UNCOV

201

    pFileState->stateFileGetFn = intervalFileGetFn;

×

UNCOV

202

    pFileState->cfName = taosStrdup("state");

×

UNCOV

203

    pFileState->stateFunctionGetFn = addRowBuffIfNotExist;

×

UNCOV

204

  } else if (type == STREAM_STATE_BUFF_SORT) {

×

UNCOV

205

    pFileState->rowStateBuff = tSimpleHashInit(cap, hashFn);

×

UNCOV

206

    pFileState->stateBuffCleanupFn = sessionWinStateCleanup;

×

UNCOV

207

    pFileState->stateBuffRemoveFn = deleteSessionWinStateBuffFn;

×

UNCOV

208

    pFileState->stateBuffRemoveByPosFn = deleteSessionWinStateBuffByPosFn;

×

UNCOV

209

    pFileState->stateBuffCreateStateKeyFn = sessionCreateStateKey;

×

UNCOV

211

    pFileState->stateFileRemoveFn = sessionFileRemoveFn;

×

UNCOV

212

    pFileState->stateFileGetFn = sessionFileGetFn;

×

UNCOV

213

    pFileState->cfName = taosStrdup("sess");

×

UNCOV

214

    pFileState->stateFunctionGetFn = getSessionRowBuff;

×

UNCOV

215

  } else if (type == STREAM_STATE_BUFF_HASH_SORT) {

×

UNCOV

216

    pFileState->rowStateBuff = tSimpleHashInit(cap, hashFn);

×

UNCOV

217

    pFileState->searchBuff = tSimpleHashInit(MIN_NUM_SEARCH_BUCKET, hashFn);

×

UNCOV

218

    QUERY_CHECK_NULL(pFileState->searchBuff, code, lino, _end, terrno);

×

UNCOV

219

    pFileState->stateBuffCleanupFn = stateHashBuffCleanupFn;

×

UNCOV

220

    pFileState->stateBuffRemoveFn = stateHashBuffRemoveFn;

×

UNCOV

221

    pFileState->stateBuffRemoveByPosFn = stateHashBuffRemoveByPosFn;

×

UNCOV

222

    pFileState->stateBuffCreateStateKeyFn = defaultCreateStateKey;

×

UNCOV

224

    pFileState->stateFileRemoveFn = hashSortFileRemoveFn;

×

UNCOV

225

    pFileState->stateFileGetFn = hashSortFileGetFn;

×

UNCOV

226

    pFileState->cfName = taosStrdup("fill");

×

UNCOV

227

    pFileState->stateFunctionGetFn = NULL;

×

UNCOV

230

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

×

UNCOV

231

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

×

UNCOV

232

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

×

UNCOV

233

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

×

UNCOV

235

  if (type == STREAM_STATE_BUFF_HASH_SEARCH) {

×

UNCOV

236

    pFileState->searchBuff = tSimpleHashInit(MIN_NUM_SEARCH_BUCKET, hashFn);

×

UNCOV

237

    QUERY_CHECK_NULL(pFileState->searchBuff, code, lino, _end, terrno);

×

UNCOV

240

  pFileState->keyLen = keySize;

×

UNCOV

241

  pFileState->rowSize = rowSize;

×

UNCOV

242

  pFileState->selectivityRowSize = selectRowSize;

×

UNCOV

243

  pFileState->preCheckPointVersion = 0;

×

UNCOV

244

  pFileState->checkPointVersion = 1;

×

UNCOV

245

  pFileState->pFileStore = pFile;

×

UNCOV

246

  pFileState->getTs = fp;

×

UNCOV

247

  pFileState->curRowCount = 0;

×

UNCOV

248

  pFileState->deleteMark = delMark;

×

UNCOV

249

  pFileState->flushMark = INT64_MIN;

×

UNCOV

250

  pFileState->maxTs = INT64_MIN;

×

UNCOV

251

  pFileState->id = taosStrdup(taskId);

×

UNCOV

252

  QUERY_CHECK_NULL(pFileState->id, code, lino, _end, terrno);

×

UNCOV

254

  pFileState->pGroupIdMap = tSimpleHashInit(1024, hashFn);

×

UNCOV

255

  QUERY_CHECK_NULL(pFileState->pGroupIdMap, code, lino, _end, terrno);

×

UNCOV

257

  pFileState->hasFillCatch = true;

×

UNCOV

259

  if (type == STREAM_STATE_BUFF_HASH || type == STREAM_STATE_BUFF_HASH_SEARCH) {

×

UNCOV

260

    code = recoverSnapshot(pFileState, checkpointId);

×

UNCOV

261

  } else if (type == STREAM_STATE_BUFF_SORT) {

×

UNCOV

262

    code = recoverSession(pFileState, checkpointId);

×

UNCOV

263

  } else if (type == STREAM_STATE_BUFF_HASH_SORT) {

×

UNCOV

264

    code = recoverFillSnapshot(pFileState, checkpointId);

×

UNCOV

266

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

268

  void*   valBuf = NULL;

×

UNCOV

269

  int32_t len = 0;

×

UNCOV

270

  int32_t tmpRes = streamDefaultGet_rocksdb(pFileState->pFileStore, STREAM_STATE_INFO_NAME, &valBuf, &len);

×

UNCOV

271

  if (tmpRes == TSDB_CODE_SUCCESS) {

×

UNCOV

272

    QUERY_CHECK_CONDITION((len == sizeof(TSKEY)), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

273

    streamFileStateDecode(&pFileState->flushMark, valBuf, len);

×

UNCOV

274

    qDebug("===stream===flushMark  read:%" PRId64, pFileState->flushMark);

×

UNCOV

276

  taosMemoryFreeClear(valBuf);

×

UNCOV

277

  (*ppFileState) = pFileState;

×

UNCOV

279

_end:

×

UNCOV

280

  if (code != TSDB_CODE_SUCCESS) {

×

281

    streamFileStateDestroy(pFileState);

×

282

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

×

UNCOV

284

  return code;

×

UNCOV

287

void destroyRowBuffPos(SRowBuffPos* pPos) {

×

UNCOV

288

  taosMemoryFreeClear(pPos->pKey);

×

UNCOV

289

  taosMemoryFreeClear(pPos->pRowBuff);

×

UNCOV

290

  taosMemoryFree(pPos);

×

UNCOV

291

×

UNCOV

293

void destroyRowBuffPosPtr(void* ptr) {

×

UNCOV

294

  if (!ptr) {

×

295

    return;

×

UNCOV

297

  SRowBuffPos* pPos = *(SRowBuffPos**)ptr;

×

UNCOV

298

  if (!pPos->beUsed) {

×

UNCOV

299

    destroyRowBuffPos(pPos);

×

UNCOV

303

void destroyRowBuffAllPosPtr(void* ptr) {

×

UNCOV

304

  if (!ptr) {

×

305

    return;

×

UNCOV

307

  SRowBuffPos* pPos = *(SRowBuffPos**)ptr;

×

UNCOV

308

  destroyRowBuffPos(pPos);

×

UNCOV

311

void destroyRowBuff(void* ptr) {

×

UNCOV

312

  if (!ptr) {

×

313

    return;

×

UNCOV

315

  taosMemoryFree(*(void**)ptr);

×

UNCOV

318

void streamFileStateDestroy(SStreamFileState* pFileState) {

×

UNCOV

319

  if (!pFileState) {

×

UNCOV

320

    return;

×

UNCOV

323

  taosMemoryFree(pFileState->id);

×

UNCOV

324

  taosMemoryFree(pFileState->cfName);

×

UNCOV

325

  tdListFreeP(pFileState->usedBuffs, destroyRowBuffAllPosPtr);

×

UNCOV

326

  tdListFreeP(pFileState->freeBuffs, destroyRowBuff);

×

UNCOV

327

  pFileState->stateBuffCleanupFn(pFileState->rowStateBuff);

×

UNCOV

328

  sessionWinStateCleanup(pFileState->searchBuff);

×

UNCOV

329

  tSimpleHashCleanup(pFileState->pGroupIdMap);

×

UNCOV

330

  taosMemoryFree(pFileState);

×

UNCOV

333

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

×

UNCOV

334

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

335

  int32_t lino = 0;

×

UNCOV

336

  if (pPos->pRowBuff) {

×

UNCOV

337

    code = tdListAppend(pFileState->freeBuffs, &(pPos->pRowBuff));

×

UNCOV

338

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

339

    pPos->pRowBuff = NULL;

×

UNCOV

342

_end:

×

UNCOV

343

  if (code != TSDB_CODE_SUCCESS) {

×

344

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

×

UNCOV

346

  return code;

×

UNCOV

349

void clearExpiredRowBuff(SStreamFileState* pFileState, TSKEY ts, bool all) {

×

UNCOV

350

  int32_t   code = TSDB_CODE_SUCCESS;

×

UNCOV

351

  int32_t   lino = 0;

×

UNCOV

352

  SListIter iter = {0};

×

UNCOV

353

  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

×

UNCOV

355

  SListNode* pNode = NULL;

×

UNCOV

356

  while ((pNode = tdListNext(&iter)) != NULL) {

×

UNCOV

357

    SRowBuffPos* pPos = *(SRowBuffPos**)(pNode->data);

×

UNCOV

358

    if (all || (pFileState->getTs(pPos->pKey) < ts && !pPos->beUsed)) {

×

UNCOV

359

      code = putFreeBuff(pFileState, pPos);

×

UNCOV

360

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

362

      if (!all) {

×

UNCOV

363

        pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

×

UNCOV

365

      destroyRowBuffPos(pPos);

×

UNCOV

366

      SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);

×

UNCOV

367

      taosMemoryFreeClear(tmp);

×

UNCOV

371

_end:

×

UNCOV

372

  if (code != TSDB_CODE_SUCCESS) {

×

373

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

×

UNCOV

375

×

UNCOV

377

int32_t clearFlushedRowBuff(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uint64_t max, bool all) {

×

UNCOV

378

  int32_t   code = TSDB_CODE_SUCCESS;

×

UNCOV

379

  int32_t   lino = 0;

×

UNCOV

380

  uint64_t  i = 0;

×

UNCOV

381

  SListIter iter = {0};

×

UNCOV

382

  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

×

UNCOV

384

  SListNode* pNode = NULL;

×

UNCOV

385

  while ((pNode = tdListNext(&iter)) != NULL && i < max) {

×

UNCOV

386

    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;

×

UNCOV

387

    if (isFlushedState(pFileState, pFileState->getTs(pPos->pKey), 0)) {

×

UNCOV

388

      if (all || !pPos->beUsed) {

×

UNCOV

389

        if (all && !pPos->pRowBuff) {

×

390

          continue;

×

UNCOV

392

        code = tdListAppend(pFlushList, &pPos);

×

UNCOV

393

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

395

        pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));

×

UNCOV

396

        pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

×

UNCOV

397

        SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);

×

UNCOV

398

        taosMemoryFreeClear(tmp);

×

UNCOV

399

        if (pPos->pRowBuff) {

×

UNCOV

400

          i++;

×

UNCOV

406

_end:

×

UNCOV

407

  if (code != TSDB_CODE_SUCCESS) {

×

408

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

×

UNCOV

410

  return code;

×

UNCOV

413

void streamFileStateClear(SStreamFileState* pFileState) {

×

UNCOV

414

  pFileState->flushMark = INT64_MIN;

×

UNCOV

415

  pFileState->maxTs = INT64_MIN;

×

UNCOV

416

  tSimpleHashClear(pFileState->rowStateBuff);

×

UNCOV

417

  clearExpiredRowBuff(pFileState, 0, true);

×

UNCOV

418

×

UNCOV

420

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

×

UNCOV

422

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

×

UNCOV

424

int32_t popUsedBuffs(SStreamFileState* pFileState, SStreamSnapshot* pFlushList, uint64_t max, bool used) {

×

UNCOV

425

  int32_t   code = TSDB_CODE_SUCCESS;

×

UNCOV

426

  int32_t   lino = 0;

×

UNCOV

427

  uint64_t  i = 0;

×

UNCOV

428

  SListIter iter = {0};

×

UNCOV

429

  tdListInitIter(pFileState->usedBuffs, &iter, TD_LIST_FORWARD);

×

UNCOV

431

  SListNode* pNode = NULL;

×

UNCOV

432

  while ((pNode = tdListNext(&iter)) != NULL && i < max) {

×

UNCOV

433

    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;

×

UNCOV

434

    if (pPos->beUsed == used) {

×

UNCOV

435

      if (used && !pPos->pRowBuff) {

×

436

        QUERY_CHECK_CONDITION((pPos->needFree == true), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

437

        continue;

×

UNCOV

439

      code = tdListAppend(pFlushList, &pPos);

×

UNCOV

440

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

442

      pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));

×

UNCOV

443

      pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

×

UNCOV

444

      SListNode* tmp = tdListPopNode(pFileState->usedBuffs, pNode);

×

UNCOV

445

      taosMemoryFreeClear(tmp);

×

UNCOV

446

      if (pPos->pRowBuff) {

×

UNCOV

447

        i++;

×

UNCOV

452

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

×

454

_end:

×

UNCOV

455

  if (code != TSDB_CODE_SUCCESS) {

×

456

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

×

UNCOV

458

  return code;

×

UNCOV

461

int32_t flushRowBuff(SStreamFileState* pFileState) {

×

UNCOV

462

  int32_t          code = TSDB_CODE_SUCCESS;

×

UNCOV

463

  int32_t          lino = 0;

×

UNCOV

464

  SStreamSnapshot* pFlushList = tdListNew(POINTER_BYTES);

×

UNCOV

465

  if (!pFlushList) {

×

466

    code = TSDB_CODE_OUT_OF_MEMORY;

×

467

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

470

  uint64_t num = (uint64_t)(pFileState->curRowCount * FLUSH_RATIO);

×

UNCOV

471

  num = TMAX(num, FLUSH_NUM);

×

UNCOV

472

  code = clearFlushedRowBuff(pFileState, pFlushList, num, false);

×

UNCOV

473

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

475

  if (isListEmpty(pFlushList)) {

×

UNCOV

476

    code = popUsedBuffs(pFileState, pFlushList, num, false);

×

UNCOV

477

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

479

    if (isListEmpty(pFlushList)) {

×

UNCOV

480

      code = popUsedBuffs(pFileState, pFlushList, num, true);

×

UNCOV

481

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

485

  if (pFileState->searchBuff) {

×

UNCOV

486

    code = clearFlushedRowBuff(pFileState, pFlushList, pFileState->curRowCount, true);

×

UNCOV

487

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

490

  flushSnapshot(pFileState, pFlushList, false);

×

UNCOV

492

  SListIter fIter = {0};

×

UNCOV

493

  tdListInitIter(pFlushList, &fIter, TD_LIST_FORWARD);

×

UNCOV

494

  SListNode* pNode = NULL;

×

UNCOV

495

  while ((pNode = tdListNext(&fIter)) != NULL) {

×

UNCOV

496

    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;

×

UNCOV

497

    code = putFreeBuff(pFileState, pPos);

×

UNCOV

498

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

501

  tdListFreeP(pFlushList, destroyRowBuffPosPtr);

×

UNCOV

503

_end:

×

UNCOV

504

  if (code != TSDB_CODE_SUCCESS) {

×

505

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

×

UNCOV

507

  return code;

×

UNCOV

510

int32_t clearRowBuff(SStreamFileState* pFileState) {

×

UNCOV

511

  if (pFileState->deleteMark != INT64_MAX) {

×

512

    clearExpiredRowBuff(pFileState, pFileState->maxTs - pFileState->deleteMark, false);

×

UNCOV

514

  if (isListEmpty(pFileState->freeBuffs)) {

×

UNCOV

515

    return flushRowBuff(pFileState);

×

UNCOV

517

  return TSDB_CODE_SUCCESS;

×

UNCOV

520

void* getFreeBuff(SStreamFileState* pFileState) {

×

UNCOV

521

  SList*     lists = pFileState->freeBuffs;

×

UNCOV

522

  int32_t    buffSize = pFileState->rowSize;

×

UNCOV

523

  SListNode* pNode = tdListPopHead(lists);

×

UNCOV

524

  if (!pNode) {

×

UNCOV

525

    return NULL;

×

UNCOV

527

  void* ptr = *(void**)pNode->data;

×

UNCOV

528

  memset(ptr, 0, buffSize);

×

UNCOV

529

  taosMemoryFree(pNode);

×

UNCOV

530

  return ptr;

×

533

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

×

534

  if (pPos->pRowBuff) {

×

535

    memset(pPos->pRowBuff, 0, pFileState->rowSize);

×

537

×

UNCOV

539

SRowBuffPos* getNewRowPos(SStreamFileState* pFileState) {

×

UNCOV

540

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

541

  int32_t      lino = 0;

×

UNCOV

542

  SRowBuffPos* pPos = taosMemoryCalloc(1, sizeof(SRowBuffPos));

×

UNCOV

543

  if (!pPos) {

×

544

    code = terrno;

×

545

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

548

  pPos->pKey = taosMemoryCalloc(1, pFileState->keyLen);

×

UNCOV

549

  if (!pPos->pKey) {

×

550

    code = terrno;

×

551

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

554

  void* pBuff = getFreeBuff(pFileState);

×

UNCOV

555

  if (pBuff) {

×

UNCOV

556

    pPos->pRowBuff = pBuff;

×

UNCOV

557

    goto _end;

×

UNCOV

560

  if (pFileState->curRowCount < pFileState->maxRowCount) {

×

UNCOV

561

    pBuff = taosMemoryCalloc(1, pFileState->rowSize);

×

UNCOV

562

    QUERY_CHECK_NULL(pBuff, code, lino, _error, terrno);

×

UNCOV

563

    pPos->pRowBuff = pBuff;

×

UNCOV

564

    pFileState->curRowCount++;

×

UNCOV

565

    goto _end;

×

UNCOV

568

  code = clearRowBuff(pFileState);

×

UNCOV

569

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

571

  pPos->pRowBuff = getFreeBuff(pFileState);

×

UNCOV

573

_end:

×

UNCOV

574

  code = tdListAppend(pFileState->usedBuffs, &pPos);

×

UNCOV

575

  QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

577

  QUERY_CHECK_CONDITION((pPos->pRowBuff != NULL), code, lino, _error, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

578

_error:

×

UNCOV

579

  if (code != TSDB_CODE_SUCCESS) {

×

580

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

×

581

    return NULL;

×

UNCOV

584

  return pPos;

×

UNCOV

587

SRowBuffPos* getNewRowPosForWrite(SStreamFileState* pFileState) {

×

UNCOV

588

  int32_t      code = TSDB_CODE_SUCCESS;

×

UNCOV

589

  int32_t      lino = 0;

×

UNCOV

590

  SRowBuffPos* newPos = getNewRowPos(pFileState);

×

UNCOV

591

  if (!newPos) {

×

592

    code = TSDB_CODE_OUT_OF_MEMORY;

×

593

    QUERY_CHECK_CODE(code, lino, _error);

×

UNCOV

595

  newPos->beUsed = true;

×

UNCOV

596

  newPos->beFlushed = false;

×

UNCOV

597

  newPos->needFree = false;

×

UNCOV

598

  newPos->beUpdated = true;

×

UNCOV

599

  return newPos;

×

601

_error:

×

602

  if (code != TSDB_CODE_SUCCESS) {

×

603

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

×

605

  return NULL;

×

UNCOV

608

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

×

UNCOV

610

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

611

  int32_t lino = 0;

×

UNCOV

612

  (*pWinCode) = TSDB_CODE_SUCCESS;

×

UNCOV

613

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

×

UNCOV

614

  SRowBuffPos** pos = tSimpleHashGet(pFileState->rowStateBuff, pKey, keyLen);

×

UNCOV

615

  if (pos) {

×

UNCOV

616

    if (pVal != NULL) {

×

UNCOV

617

      *pVLen = pFileState->rowSize;

×

UNCOV

618

      *pVal = *pos;

×

UNCOV

619

      (*pos)->beUsed = true;

×

UNCOV

620

      (*pos)->beFlushed = false;

×

UNCOV

622

    goto _end;

×

UNCOV

624

  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

UNCOV

625

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

×

626

    code = TSDB_CODE_OUT_OF_MEMORY;

×

627

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

630

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

×

UNCOV

631

  (*pWinCode) = TSDB_CODE_FAILED;

×

UNCOV

633

  TSKEY ts = pFileState->getTs(pKey);

×

UNCOV

634

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

×

UNCOV

635

    int32_t len = 0;

×

UNCOV

636

    void*   p = NULL;

×

UNCOV

637

    (*pWinCode) = pFileState->stateFileGetFn(pFileState, pKey, &p, &len);

×

UNCOV

638

    qDebug("===stream===get %" PRId64 " from disc, res %d", ts, (*pWinCode));

×

UNCOV

639

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

UNCOV

640

      memcpy(pNewPos->pRowBuff, p, len);

×

UNCOV

642

    taosMemoryFree(p);

×

UNCOV

645

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

×

UNCOV

646

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

648

  if (pVal) {

×

UNCOV

649

    *pVLen = pFileState->rowSize;

×

UNCOV

650

    *pVal = pNewPos;

×

UNCOV

653

_end:

×

UNCOV

654

  if (code != TSDB_CODE_SUCCESS) {

×

655

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

×

UNCOV

657

  return code;

×

UNCOV

660

void deleteRowBuff(SStreamFileState* pFileState, const void* pKey, int32_t keyLen) {

×

UNCOV

661

  int32_t code_buff = pFileState->stateBuffRemoveFn(pFileState->rowStateBuff, pKey, keyLen);

×

UNCOV

662

  qTrace("%s at line %d res:%d", __func__, __LINE__, code_buff);

×

UNCOV

663

  int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey);

×

UNCOV

664

  qTrace("%s at line %d res:%d", __func__, __LINE__, code_file);

×

UNCOV

665

  if (pFileState->searchBuff != NULL) {

×

UNCOV

666

    deleteHashSortRowBuff(pFileState, pKey);

×

UNCOV

668

×

NEW

670

void deleteRowBuffByGroupId(SStreamFileState* pFileState, uint64_t groupId) {

×

NEW

671

  SSHashObj* pRowMap = pFileState->rowStateBuff;

×

NEW

672

  void*   pIte = NULL;

×

NEW

673

  int32_t iter = 0;

×

NEW

674

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

×

NEW

675

    size_t keyLen = 0;

×

NEW

676

    SWinKey* pKey = tSimpleHashGetKey(pIte, &keyLen);

×

NEW

677

    if (pKey->groupId == groupId) {

×

NEW

678

      int32_t tmpRes = tSimpleHashIterateRemove(pRowMap, pKey, keyLen, &pIte, &iter);

×

NEW

679

      qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);

×

NEW

683

  while (1) {

×

NEW

684

    SWinKey tmp = {.ts = INT64_MIN, .groupId = groupId};

×

NEW

685

    SStreamStateCur* pCur = streamStateSeekKeyNext_rocksdb(pFileState->pFileStore, &tmp);

×

NEW

686

    SWinKey delKey = {.groupId = groupId};

×

NEW

687

    int32_t code = streamStateGetGroupKVByCur_rocksdb(pFileState->pFileStore, pCur, &delKey, NULL, 0);

×

NEW

688

    if (code != TSDB_CODE_SUCCESS) {

×

NEW

689

      break;

×

NEW

691

    code = streamStateDel_rocksdb(pFileState->pFileStore, &delKey);

×

NEW

692

    qTrace("%s at line %d res:%d", __func__, __LINE__, code);

×

NEW

694

×

UNCOV

696

static int32_t recoverSessionRowBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {

×

UNCOV

697

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

698

  int32_t lino = 0;

×

UNCOV

699

  int32_t len = 0;

×

UNCOV

700

  void*   pBuff = NULL;

×

UNCOV

701

  code = pFileState->stateFileGetFn(pFileState, pPos->pKey, &pBuff, &len);

×

UNCOV

702

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

703

  memcpy(pPos->pRowBuff, pBuff, len);

×

UNCOV

704

  taosMemoryFree(pBuff);

×

UNCOV

706

_end:

×

UNCOV

707

  if (code != TSDB_CODE_SUCCESS) {

×

708

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

×

UNCOV

710

  return code;

×

UNCOV

713

static int32_t recoverStateRowBuff(SStreamFileState* pFileState, SRowBuffPos* pPos) {

×

UNCOV

714

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

715

  int32_t lino = 0;

×

UNCOV

716

  pPos->pRowBuff = getFreeBuff(pFileState);

×

UNCOV

717

  if (!pPos->pRowBuff) {

×

UNCOV

718

    if (pFileState->curRowCount < pFileState->maxRowCount) {

×

719

      pPos->pRowBuff = taosMemoryCalloc(1, pFileState->rowSize);

×

720

      if (!pPos->pRowBuff) {

×

721

        code = terrno;

×

722

        QUERY_CHECK_CODE(code, lino, _end);

×

724

      pFileState->curRowCount++;

×

UNCOV

726

      code = clearRowBuff(pFileState);

×

UNCOV

727

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

728

      pPos->pRowBuff = getFreeBuff(pFileState);

×

UNCOV

730

    QUERY_CHECK_CONDITION((pPos->pRowBuff != NULL), code, lino, _end, TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR);

×

UNCOV

733

  code = recoverSessionRowBuff(pFileState, pPos);

×

UNCOV

734

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

736

_end:

×

UNCOV

737

  if (code != TSDB_CODE_SUCCESS) {

×

738

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

×

UNCOV

740

  return code;

×

UNCOV

743

int32_t getRowBuffByPos(SStreamFileState* pFileState, SRowBuffPos* pPos, void** pVal) {

×

UNCOV

744

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

745

  int32_t lino = 0;

×

UNCOV

746

  if (pPos->pRowBuff) {

×

UNCOV

747

    if (pPos->needFree) {

×

748

      code = recoverSessionRowBuff(pFileState, pPos);

×

749

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

751

    (*pVal) = pPos->pRowBuff;

×

UNCOV

752

    goto _end;

×

UNCOV

755

  code = recoverStateRowBuff(pFileState, pPos);

×

UNCOV

756

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

758

  (*pVal) = pPos->pRowBuff;

×

UNCOV

759

  if (!pPos->needFree) {

×

UNCOV

760

    code = tdListPrepend(pFileState->usedBuffs, &pPos);

×

UNCOV

761

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

764

_end:

×

UNCOV

765

  if (code != TSDB_CODE_SUCCESS) {

×

766

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

×

UNCOV

768

  return code;

×

UNCOV

771

bool hasRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen) {

×

UNCOV

772

  SRowBuffPos** pos = tSimpleHashGet(pFileState->rowStateBuff, pKey, keyLen);

×

UNCOV

773

  if (pos) {

×

UNCOV

774

    return true;

×

UNCOV

776

  return false;

×

UNCOV

779

SStreamSnapshot* getSnapshot(SStreamFileState* pFileState) {

×

UNCOV

780

  int64_t mark = (pFileState->deleteMark == INT64_MAX || pFileState->maxTs == INT64_MIN)

×

UNCOV

782

                     : pFileState->maxTs - pFileState->deleteMark;

×

UNCOV

783

  clearExpiredRowBuff(pFileState, mark, false);

×

UNCOV

784

  return pFileState->usedBuffs;

×

UNCOV

787

void flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot, bool flushState) {

×

UNCOV

788

  int32_t   code = TSDB_CODE_SUCCESS;

×

UNCOV

789

  int32_t   lino = 0;

×

UNCOV

790

  SListIter iter = {0};

×

UNCOV

791

  tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD);

×

UNCOV

793

  const int32_t BATCH_LIMIT = 256;

×

UNCOV

795

  int64_t    st = taosGetTimestampMs();

×

UNCOV

796

  SListNode* pNode = NULL;

×

UNCOV

798

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

×

UNCOV

800

  int32_t len = (pFileState->rowSize + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;

×

UNCOV

801

  char*   buf = taosMemoryCalloc(1, len);

×

UNCOV

802

  if (!buf) {

×

803

    code = terrno;

×

804

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

807

  void* batch = streamStateCreateBatch();

×

UNCOV

808

  if (!batch) {

×

809

    code = TSDB_CODE_OUT_OF_MEMORY;

×

810

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

813

  while ((pNode = tdListNext(&iter)) != NULL && code == TSDB_CODE_SUCCESS) {

×

UNCOV

814

    SRowBuffPos* pPos = *(SRowBuffPos**)pNode->data;

×

UNCOV

815

    if (pPos->beFlushed || !pPos->pRowBuff) {

×

UNCOV

816

      continue;

×

UNCOV

818

    pPos->beFlushed = true;

×

UNCOV

819

    pFileState->flushMark = TMAX(pFileState->flushMark, pFileState->getTs(pPos->pKey));

×

UNCOV

821

    qDebug("===stream===flushed start:%" PRId64, pFileState->getTs(pPos->pKey));

×

UNCOV

822

    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {

×

UNCOV

823

      code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);

×

UNCOV

824

      streamStateClearBatch(batch);

×

UNCOV

825

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

828

    void* pSKey = pFileState->stateBuffCreateStateKeyFn(pPos, ((SStreamState*)pFileState->pFileStore)->number);

×

UNCOV

829

    QUERY_CHECK_NULL(pSKey, code, lino, _end, terrno);

×

UNCOV

831

    code = streamStatePutBatchOptimize(pFileState->pFileStore, idx, batch, pSKey, pPos->pRowBuff, pFileState->rowSize,

×

UNCOV

833

    taosMemoryFreeClear(pSKey);

×

UNCOV

834

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

836

    memset(buf, 0, len);

×

UNCOV

838

  taosMemoryFreeClear(buf);

×

UNCOV

840

  int32_t numOfElems = streamStateGetBatchSize(batch);

×

UNCOV

841

  if (numOfElems > 0) {

×

UNCOV

842

    code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);

×

UNCOV

843

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

845

    goto _end;

×

UNCOV

848

  streamStateClearBatch(batch);

×

UNCOV

850

  clearSearchBuff(pFileState);

×

UNCOV

852

  int64_t elapsed = taosGetTimestampMs() - st;

×

UNCOV

853

  qDebug("%s flush to disk in batch model completed, rows:%d, batch size:%d, elapsed time:%" PRId64 "ms",

×

UNCOV

856

  if (flushState) {

×

UNCOV

857

    void*   valBuf = NULL;

×

UNCOV

858

    int32_t len = 0;

×

UNCOV

859

    code = streamFileStateEncode(&pFileState->flushMark, &valBuf, &len);

×

UNCOV

860

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

862

    qDebug("===stream===flushMark write:%" PRId64, pFileState->flushMark);

×

UNCOV

863

    code = streamStatePutBatch(pFileState->pFileStore, "default", batch, STREAM_STATE_INFO_NAME, valBuf, len, 0);

×

UNCOV

864

    taosMemoryFree(valBuf);

×

UNCOV

865

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

867

    code = streamStatePutBatch_rocksdb(pFileState->pFileStore, batch);

×

UNCOV

868

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

871

_end:

×

UNCOV

872

  if (code != TSDB_CODE_SUCCESS) {

×

873

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

×

UNCOV

875

  taosMemoryFree(buf);

×

UNCOV

876

  streamStateDestroyBatch(batch);

×

UNCOV

877

×

879

int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {

×

880

  char keyBuf[128] = {0};

×

881

  TAOS_UNUSED(tsnprintf(keyBuf, sizeof(keyBuf), "%s:%" PRId64 "", TASK_KEY, checkpointId));

×

882

  return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);

×

885

int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {

×

886

  int32_t code = TSDB_CODE_SUCCESS;

×

887

  int64_t maxCheckPointId = 0;

×

889

    char    buf[128] = {0};

×

890

    void*   val = NULL;

×

891

    int32_t len = 0;

×

892

    memcpy(buf, TASK_KEY, strlen(TASK_KEY));

×

893

    code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);

×

894

    if (code != 0 || len == 0 || val == NULL) {

×

895

      return TSDB_CODE_FAILED;

×

897

    memcpy(buf, val, len);

×

898

    buf[len] = 0;

×

899

    maxCheckPointId = taosStr2Int64((char*)buf, NULL, 10);

×

900

    taosMemoryFree(val);

×

902

  for (int64_t i = maxCheckPointId; i > 0; i--) {

×

903

    char    buf[128] = {0};

×

904

    void*   val = 0;

×

905

    int32_t len = 0;

×

906

    TAOS_UNUSED(tsnprintf(buf, sizeof(buf), "%s:%" PRId64 "", TASK_KEY, i));

×

907

    code = streamDefaultGet_rocksdb(pFileState->pFileStore, buf, &val, &len);

×

908

    if (code != 0) {

×

909

      return TSDB_CODE_FAILED;

×

911

    memcpy(buf, val, len);

×

912

    buf[len] = 0;

×

913

    taosMemoryFree(val);

×

916

    ts = taosStr2Int64((char*)buf, NULL, 10);

×

917

    if (ts < mark) {

×

919

      int32_t tmpRes = forceRemoveCheckpoint(pFileState, i);

×

920

      qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);

×

921

      break;

×

924

  return code;

×

UNCOV

927

int32_t recoverSession(SStreamFileState* pFileState, int64_t ckId) {

×

UNCOV

928

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

929

  int32_t lino = 0;

×

UNCOV

930

  int32_t winRes = TSDB_CODE_SUCCESS;

×

UNCOV

931

  if (pFileState->maxTs != INT64_MIN) {

×

932

    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)

×

934

                       : pFileState->maxTs - pFileState->deleteMark;

×

935

    int32_t tmpRes = deleteExpiredCheckPoint(pFileState, mark);

×

936

    qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);

×

UNCOV

939

  SStreamStateCur* pCur = streamStateSessionSeekToLast_rocksdb(pFileState->pFileStore, INT64_MAX);

×

UNCOV

940

  int32_t          recoverNum = TMIN(MIN_NUM_OF_RECOVER_ROW_BUFF, pFileState->maxRowCount);

×

UNCOV

941

  while (winRes == TSDB_CODE_SUCCESS) {

×

UNCOV

942

    if (pFileState->curRowCount >= recoverNum) {

×

UNCOV

943

      break;

×

UNCOV

946

    void*       pVal = NULL;

×

UNCOV

947

    int32_t     vlen = 0;

×

UNCOV

948

    SSessionKey key = {0};

×

UNCOV

949

    winRes = streamStateSessionGetKVByCur_rocksdb(getStateFileStore(pFileState), pCur, &key, &pVal, &vlen);

×

UNCOV

950

    if (winRes != TSDB_CODE_SUCCESS) {

×

UNCOV

951

      break;

×

954

    if (vlen != pFileState->rowSize) {

×

955

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

956

      QUERY_CHECK_CODE(code, lino, _end);

×

959

    SRowBuffPos* pPos = createSessionWinBuff(pFileState, &key, pVal, &vlen);

×

960

    pPos->beUsed = false;

×

961

    winRes = putSessionWinResultBuff(pFileState, pPos);

×

962

    if (winRes != TSDB_CODE_SUCCESS) {

×

963

      break;

×

966

    winRes = streamStateSessionCurPrev_rocksdb(pCur);

×

969

_end:

×

UNCOV

970

  if (code != TSDB_CODE_SUCCESS) {

×

971

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

×

UNCOV

973

  streamStateFreeCur(pCur);

×

UNCOV

974

  return code;

×

UNCOV

977

int32_t recoverSnapshot(SStreamFileState* pFileState, int64_t ckId) {

×

UNCOV

978

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

979

  int32_t lino = 0;

×

UNCOV

980

  int32_t winCode = TSDB_CODE_SUCCESS;

×

UNCOV

981

  if (pFileState->maxTs != INT64_MIN) {

×

982

    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)

×

984

                       : pFileState->maxTs - pFileState->deleteMark;

×

985

    int32_t tmpRes = deleteExpiredCheckPoint(pFileState, mark);

×

986

    qTrace("%s at line %d res:%d", __func__, __LINE__, tmpRes);

×

UNCOV

989

  SStreamStateCur* pCur = streamStateSeekToLast_rocksdb(pFileState->pFileStore);

×

UNCOV

990

  int32_t          recoverNum = TMIN(MIN_NUM_OF_RECOVER_ROW_BUFF, pFileState->maxRowCount);

×

UNCOV

991

  while (winCode == TSDB_CODE_SUCCESS) {

×

UNCOV

992

    if (pFileState->curRowCount >= recoverNum) {

×

UNCOV

993

      break;

×

UNCOV

996

    void*        pVal = NULL;

×

UNCOV

997

    int32_t      vlen = 0;

×

UNCOV

998

    SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

UNCOV

999

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

×

1000

      code = TSDB_CODE_OUT_OF_MEMORY;

×

1001

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1004

    winCode = streamStateGetKVByCur_rocksdb(getStateFileStore(pFileState), pCur, pNewPos->pKey, (const void**)&pVal, &vlen);

×

UNCOV

1005

    qDebug("===stream=== get state by cur winres:%d. %s", winCode, __func__);

×

UNCOV

1006

    if (winCode != TSDB_CODE_SUCCESS || pFileState->getTs(pNewPos->pKey) < pFileState->flushMark) {

×

UNCOV

1007

      destroyRowBuffPos(pNewPos);

×

UNCOV

1008

      SListNode* pNode = tdListPopTail(pFileState->usedBuffs);

×

UNCOV

1009

      taosMemoryFreeClear(pNode);

×

UNCOV

1010

      taosMemoryFreeClear(pVal);

×

UNCOV

1011

      break;

×

1013

    if (vlen != pFileState->rowSize) {

×

1014

      qError("row size mismatch, expect:%d, actual:%d", pFileState->rowSize, vlen);

×

1015

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1016

      taosMemoryFreeClear(pVal);

×

1017

      QUERY_CHECK_CODE(code, lino, _end);

×

1019

    memcpy(pNewPos->pRowBuff, pVal, vlen);

×

1020

    taosMemoryFreeClear(pVal);

×

1021

    pNewPos->beFlushed = true;

×

1022

    pNewPos->beUsed = false;

×

1023

    qDebug("===stream=== read checkpoint state from disc. %s", __func__);

×

1024

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

×

1025

    if (code != TSDB_CODE_SUCCESS) {

×

1026

      destroyRowBuffPos(pNewPos);

×

1027

      break;

×

1029

    streamStateCurPrev_rocksdb(pCur);

×

1032

_end:

×

UNCOV

1033

  if (code != TSDB_CODE_SUCCESS) {

×

1034

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

×

UNCOV

1036

  streamStateFreeCur(pCur);

×

UNCOV

1037

  return code;

×

UNCOV

1040

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

×

UNCOV

1042

void streamFileStateReloadInfo(SStreamFileState* pFileState, TSKEY ts) {

×

UNCOV

1043

  pFileState->flushMark = TMAX(pFileState->flushMark, ts);

×

UNCOV

1044

  pFileState->maxTs = TMAX(pFileState->maxTs, ts);

×

UNCOV

1045

×

UNCOV

1047

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

×

UNCOV

1048

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

×

UNCOV

1050

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

×

UNCOV

1052

bool isDeteled(SStreamFileState* pFileState, TSKEY ts) {

×

UNCOV

1053

  return pFileState->deleteMark != INT64_MAX && pFileState->maxTs > 0 &&

×

UNCOV

1054

         ts < (pFileState->maxTs - pFileState->deleteMark);

×

UNCOV

1057

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

×

UNCOV

1059

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

×

UNCOV

1061

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

×

UNCOV

1063

int32_t getFunctionRowBuff(SStreamFileState* pFileState, void* pKey, int32_t keyLen, void** pVal, int32_t* pVLen) {

×

UNCOV

1064

  int32_t winCode = TSDB_CODE_SUCCESS;

×

UNCOV

1065

  return pFileState->stateFunctionGetFn(pFileState, pKey, keyLen, pVal, pVLen, &winCode);

×

UNCOV

1068

int32_t recoverFillSnapshot(SStreamFileState* pFileState, int64_t ckId) {

×

UNCOV

1069

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1070

  int32_t lino = 0;

×

UNCOV

1071

  if (pFileState->maxTs != INT64_MIN) {

×

1072

    int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)

×

1074

                       : pFileState->maxTs - pFileState->deleteMark;

×

1075

    code = deleteExpiredCheckPoint(pFileState, mark);

×

1076

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1079

  SStreamStateCur* pCur = streamStateFillSeekToLast_rocksdb(pFileState->pFileStore);

×

UNCOV

1080

  if (pCur == NULL) {

×

UNCOV

1081

    return code;

×

UNCOV

1083

  int32_t recoverNum = TMIN(MIN_NUM_OF_RECOVER_ROW_BUFF, pFileState->maxRowCount);

×

UNCOV

1084

  int32_t winRes = TSDB_CODE_SUCCESS;

×

UNCOV

1085

  while (winRes == TSDB_CODE_SUCCESS) {

×

UNCOV

1086

    if (pFileState->curRowCount >= recoverNum) {

×

UNCOV

1087

      break;

×

UNCOV

1090

    void*        pVal = NULL;

×

UNCOV

1091

    int32_t      vlen = 0;

×

UNCOV

1092

    SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

UNCOV

1093

    winRes = streamStateFillGetKVByCur_rocksdb(pCur, pNewPos->pKey, (const void**)&pVal, &vlen);

×

UNCOV

1094

    qDebug("===stream=== get state by cur winres:%d. %s", winRes, __func__);

×

UNCOV

1095

    if (winRes != TSDB_CODE_SUCCESS || isFlushedState(pFileState, pFileState->getTs(pNewPos->pKey), 0)) {

×

UNCOV

1096

      destroyRowBuffPos(pNewPos);

×

UNCOV

1097

      SListNode* pNode = tdListPopTail(pFileState->usedBuffs);

×

UNCOV

1098

      taosMemoryFreeClear(pNode);

×

UNCOV

1099

      taosMemoryFreeClear(pVal);

×

UNCOV

1100

      break;

×

UNCOV

1103

    if (vlen != pFileState->rowSize) {

×

1104

      qError("row size mismatch, expect:%d, actual:%d", pFileState->rowSize, vlen);

×

1105

      destroyRowBuffPos(pNewPos);

×

1106

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1107

      taosMemoryFreeClear(pVal);

×

1108

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1111

    memcpy(pNewPos->pRowBuff, pVal, vlen);

×

UNCOV

1112

    taosMemoryFreeClear(pVal);

×

UNCOV

1113

    pNewPos->beFlushed = true;

×

UNCOV

1114

    pNewPos->beUsed = false;

×

UNCOV

1115

    qDebug("===stream=== read checkpoint state from disc. %s", __func__);

×

UNCOV

1116

    winRes = tSimpleHashPut(pFileState->rowStateBuff, pNewPos->pKey, pFileState->keyLen, &pNewPos, POINTER_BYTES);

×

UNCOV

1117

    if (winRes != TSDB_CODE_SUCCESS) {

×

1118

      destroyRowBuffPos(pNewPos);

×

1119

      break;

×

UNCOV

1121

    streamStateCurPrev_rocksdb(pCur);

×

UNCOV

1123

  streamStateFreeCur(pCur);

×

UNCOV

1125

_end:

×

UNCOV

1126

  if (code != TSDB_CODE_SUCCESS) {

×

1127

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

×

UNCOV

1129

  return code;

×

UNCOV

1132

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

×

UNCOV

1134

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1135

  int32_t lino = 0;

×

UNCOV

1136

  (*pWinCode) = TSDB_CODE_FAILED;

×

UNCOV

1137

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

×

UNCOV

1138

  SRowBuffPos** ppPos = tSimpleHashGet(pFileState->rowStateBuff, pKey, keyLen);

×

UNCOV

1139

  if (ppPos) {

×

UNCOV

1140

    *pVLen = pFileState->rowSize;

×

UNCOV

1141

    *pVal = *ppPos;

×

UNCOV

1142

    (*ppPos)->beUsed = true;

×

UNCOV

1143

    (*ppPos)->beFlushed = false;

×

UNCOV

1144

    (*pWinCode) = TSDB_CODE_SUCCESS;

×

UNCOV

1145

    if ((*ppPos)->pRowBuff == NULL) {

×

1146

      code = recoverStateRowBuff(pFileState, *ppPos);

×

1147

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1149

    goto _end;

×

UNCOV

1151

  TSKEY ts = pFileState->getTs(pKey);

×

UNCOV

1152

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

×

UNCOV

1153

    int32_t len = 0;

×

UNCOV

1154

    void*   p = NULL;

×

UNCOV

1155

    (*pWinCode) = pFileState->stateFileGetFn(pFileState, pKey, &p, &len);

×

UNCOV

1156

    qDebug("===stream===get %" PRId64 " from disc, res %d", ts, (*pWinCode));

×

UNCOV

1157

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

UNCOV

1158

      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

UNCOV

1159

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

×

1160

        code = TSDB_CODE_OUT_OF_MEMORY;

×

1161

        QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1164

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

×

UNCOV

1165

      memcpy(pNewPos->pRowBuff, p, len);

×

UNCOV

1166

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

×

UNCOV

1167

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1169

      if (pVal) {

×

UNCOV

1170

        *pVLen = pFileState->rowSize;

×

UNCOV

1171

        *pVal = pNewPos;

×

UNCOV

1174

    taosMemoryFree(p);

×

UNCOV

1177

_end:

×

UNCOV

1178

  if (code != TSDB_CODE_SUCCESS) {

×

1179

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

×

UNCOV

1181

  return code;

×

UNCOV

1184

int32_t streamFileStateGroupPut(SStreamFileState* pFileState, int64_t groupId, void* value, int32_t vLen) {

×

UNCOV

1185

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1186

  int32_t lino = 0;

×

UNCOV

1187

  if (value != NULL) {

×

1188

    code = TSDB_CODE_INVALID_PARA;

×

1189

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1192

  if (tSimpleHashGet(pFileState->pGroupIdMap, &groupId, sizeof(int64_t)) == NULL) {

×

UNCOV

1193

    if (tSimpleHashGetSize(pFileState->pGroupIdMap) <= MAX_GROUP_ID_NUM) {

×

UNCOV

1194

      code = tSimpleHashPut(pFileState->pGroupIdMap, &groupId, sizeof(int64_t), NULL, 0);

×

UNCOV

1195

      QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1197

    code = streamStatePutParTag_rocksdb(pFileState->pFileStore, groupId, value, vLen);

×

UNCOV

1198

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1201

_end:

×

UNCOV

1202

  if (code != TSDB_CODE_SUCCESS) {

×

1203

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

×

UNCOV

1205

  return code;

×

UNCOV

1208

void streamFileStateGroupCurNext(SStreamStateCur* pCur) {

×

UNCOV

1209

  SStreamFileState* pFileState = (SStreamFileState*)pCur->pStreamFileState;

×

UNCOV

1210

  if (pCur->hashIter == -1) {

×

UNCOV

1211

    streamStateCurNext(pFileState->pFileStore, pCur);

×

UNCOV

1212

    return;

×

UNCOV

1215

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

×

UNCOV

1216

  pCur->minGpId = TMAX(pCur->minGpId, gpId);

×

UNCOV

1218

  SSHashObj* pHash = pFileState->pGroupIdMap;

×

UNCOV

1219

  pCur->pHashData = tSimpleHashIterate(pHash, pCur->pHashData, &pCur->hashIter);

×

UNCOV

1220

  if (!pCur->pHashData) {

×

UNCOV

1221

    pCur->hashIter = -1;

×

UNCOV

1222

    streamStateParTagSeekKeyNext_rocksdb(pFileState->pFileStore, pCur->minGpId, pCur);

×

UNCOV

1223

    return;

×

UNCOV

1227

int32_t streamFileStateGroupGetKVByCur(SStreamStateCur* pCur, int64_t* pKey, void** pVal, int32_t* pVLen) {

×

UNCOV

1228

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1229

  if (pCur->pHashData) {

×

UNCOV

1230

    *pKey = *(int64_t*)tSimpleHashGetKey(pCur->pHashData, NULL);

×

UNCOV

1231

    return code;

×

UNCOV

1233

  return streamStateParTagGetKVByCur_rocksdb(pCur, pKey, NULL, NULL);

×

UNCOV

1236

SSHashObj* getGroupIdCache(SStreamFileState* pFileState) {

×

UNCOV

1237

  return pFileState->pGroupIdMap;

×

UNCOV

1240

void clearExpiredState(SStreamFileState* pFileState) {

×

UNCOV

1241

  int32_t    code = TSDB_CODE_SUCCESS;

×

UNCOV

1242

  int32_t    lino = 0;

×

UNCOV

1243

  SSHashObj* pSearchBuff = pFileState->searchBuff;

×

UNCOV

1244

  void*      pIte = NULL;

×

UNCOV

1245

  int32_t    iter = 0;

×

UNCOV

1246

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

×

UNCOV

1247

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

×

UNCOV

1248

    int32_t size = taosArrayGetSize(pWinStates);

×

UNCOV

1249

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

×

UNCOV

1250

      SWinKey* pKey = taosArrayGet(pWinStates, i);

×

UNCOV

1251

      int32_t  code_buff = pFileState->stateBuffRemoveFn(pFileState->rowStateBuff, pKey, sizeof(SWinKey));

×

UNCOV

1252

      qTrace("clear expired buff, ts:%" PRId64 ". %s at line %d res:%d", pKey->ts, __func__, __LINE__, code_buff);

×

UNCOV

1254

      if (isFlushedState(pFileState, pKey->ts, 0)) {

×

UNCOV

1255

        int32_t code_file = pFileState->stateFileRemoveFn(pFileState, pKey);

×

UNCOV

1256

        qTrace("clear expired file, ts:%" PRId64 ". %s at line %d res:%d", pKey->ts, __func__, __LINE__, code_file);

×

UNCOV

1259

    taosArrayRemoveBatch(pWinStates, 0, size - 1, NULL);

×

UNCOV

1261

  code = clearRowBuff(pFileState);

×

UNCOV

1262

  QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1264

_end:

×

UNCOV

1265

  if (code != TSDB_CODE_SUCCESS) {

×

1266

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

×

UNCOV

1268

×

UNCOV

1340

int32_t getRowStatePrevRow(SStreamFileState* pFileState, const SWinKey* pKey, SWinKey* pResKey, void** ppVal,

×

UNCOV

1342

  int32_t    code = TSDB_CODE_SUCCESS;

×

UNCOV

1343

  int32_t    lino = 0;

×

UNCOV

1344

  SArray*    pWinStates = NULL;

×

UNCOV

1345

  SSHashObj* pSearchBuff = getSearchBuff(pFileState);

×

UNCOV

1346

  void*      pState = getStateFileStore(pFileState);

×

UNCOV

1347

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

×

UNCOV

1348

  if (ppBuff) {

×

UNCOV

1349

    pWinStates = (SArray*)(*ppBuff);

×

1351

    qDebug("===stream=== search buff is empty.group id:%" PRId64, pKey->groupId);

×

1352

    SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, pKey);

×

1353

    void*            tmpVal = NULL;

×

1354

    int32_t          len = 0;

×

1355

    (*pWinCode) = streamStateGetGroupKVByCur_rocksdb(pState, pCur, pResKey, (const void**)&tmpVal, &len);

×

1356

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

1357

      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1358

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

×

1359

        code = TSDB_CODE_OUT_OF_MEMORY;

×

1360

        QUERY_CHECK_CODE(code, lino, _end);

×

1362

      memcpy(pNewPos->pRowBuff, tmpVal, len);

×

1363

      taosMemoryFreeClear(tmpVal);

×

1364

      *pVLen = getRowStateRowSize(pFileState);

×

1365

      (*ppVal) = pNewPos;

×

1367

    streamStateFreeCur(pCur);

×

1368

    return code;

×

UNCOV

1370

  int32_t size = taosArrayGetSize(pWinStates);

×

UNCOV

1371

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

×

UNCOV

1372

  if (index >= 0) {

×

UNCOV

1373

    SWinKey* pCurKey = taosArrayGet(pWinStates, index);

×

UNCOV

1374

    if (winKeyCmprImpl(pCurKey, pKey) == 0) {

×

UNCOV

1375

      index--;

×

1377

      qDebug("%s failed at line %d since do not find cur SWinKey. trigger may be force window close", __func__, __LINE__);

×

UNCOV

1380

  if (index == -1) {

×

UNCOV

1381

    SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, pKey);

×

UNCOV

1382

    void*            tmpVal = NULL;

×

UNCOV

1383

    int32_t          len = 0;

×

UNCOV

1384

    (*pWinCode) = streamStateGetGroupKVByCur_rocksdb(pState, pCur, pResKey, (const void**)&tmpVal, &len);

×

UNCOV

1385

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

1386

      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1387

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

×

1388

        code = TSDB_CODE_OUT_OF_MEMORY;

×

1389

        QUERY_CHECK_CODE(code, lino, _end);

×

1391

      memcpy(pNewPos->pRowBuff, tmpVal, len);

×

1392

      taosMemoryFreeClear(tmpVal);

×

1393

      *pVLen = getRowStateRowSize(pFileState);

×

1394

      (*ppVal) = pNewPos;

×

UNCOV

1396

    streamStateFreeCur(pCur);

×

UNCOV

1397

    return code;

×

UNCOV

1399

    SWinKey* pPrevKey = taosArrayGet(pWinStates, index);

×

UNCOV

1400

    *pResKey = *pPrevKey;

×

UNCOV

1401

    return addRowBuffIfNotExist(pFileState, (void*)pPrevKey, sizeof(SWinKey), ppVal, pVLen, pWinCode);

×

1405

_end:

×

1406

  if (code != TSDB_CODE_SUCCESS) {

×

1407

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

×

1409

  return code;

×

UNCOV

1412

int32_t addSearchItem(SStreamFileState* pFileState, SArray* pWinStates, const SWinKey* pKey) {

×

UNCOV

1413

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1414

  int32_t lino = 0;

×

UNCOV

1415

  int32_t size = taosArrayGetSize(pWinStates);

×

UNCOV

1416

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

×

UNCOV

1417

  if (!isFlushedState(pFileState, pKey->ts, 0) || index >= 0 || size == 0) {

×

UNCOV

1418

    if (index >= 0) {

×

UNCOV

1419

      SWinKey* pTmpKey = taosArrayGet(pWinStates, index);

×

UNCOV

1420

      if (winKeyCmprImpl(pTmpKey, pKey) == 0) {

×

UNCOV

1421

        goto _end;

×

UNCOV

1424

    index++;

×

UNCOV

1425

    void* tmp = taosArrayInsert(pWinStates, index, pKey);

×

UNCOV

1426

    QUERY_CHECK_NULL(tmp, code, lino, _end, terrno);

×

UNCOV

1429

  if (size >= MAX_NUM_OF_CACHE_WIN) {

×

1430

    int32_t num = size - NUM_OF_CACHE_WIN;

×

1431

    taosArrayRemoveBatch(pWinStates, 0, num, NULL);

×

UNCOV

1433

_end:

×

UNCOV

1434

  if (code != TSDB_CODE_SUCCESS) {

×

1435

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

×

UNCOV

1437

  return code;

×

UNCOV

1440

int32_t addArrayBuffIfNotExist(SSHashObj* pSearchBuff, uint64_t groupId, SArray** ppResStates) {

×

UNCOV

1441

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

1442

  int32_t lino = 0;

×

UNCOV

1443

  SArray*    pWinStates = NULL;

×

UNCOV

1444

  void**     ppBuff = tSimpleHashGet(pSearchBuff, &groupId, sizeof(uint64_t));

×

UNCOV

1445

  if (ppBuff) {

×

UNCOV

1446

    pWinStates = (SArray*)(*ppBuff);

×

UNCOV

1448

    pWinStates = taosArrayInit(16, sizeof(SWinKey));

×

UNCOV

1449

    QUERY_CHECK_NULL(pWinStates, code, lino, _end, terrno);

×

UNCOV

1451

    code = tSimpleHashPut(pSearchBuff, &groupId, sizeof(uint64_t), &pWinStates, POINTER_BYTES);

×

UNCOV

1452

    QUERY_CHECK_CODE(code, lino, _end);

×

UNCOV

1455

  (*ppResStates) = pWinStates;

×

UNCOV

1457

_end:

×

UNCOV

1458

  if (code != TSDB_CODE_SUCCESS) {

×

1459

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

×

UNCOV

1461

  return code;

×

taosdata / TDengine / #3621

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

taosdata / TDengine / #3621

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