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

85

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

×

86

  SWinKey* pWin2 = taosArrayGet(pDatas, pos);

×

87

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

×

90

int fillTSKeyCompare(const void* pKey1, const void* pDatas, int pos) {

×

91

  SWinKey* pWin1 = (SWinKey*)pKey1;

×

92

  SWinKey* pWin2 = taosArrayGet(pDatas, pos);

×

93

  if (pWin1->ts > pWin2->ts) {

×

94

    return 1;

×

95

  } else if (pWin1->ts < pWin2->ts) {

×

96

    return -1;

×

99

  return 0;

×

102

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

×

103

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

×

104

  if (pos) {

×

105

    (*pos)->beFlushed = true;

×

106

    (*pos)->invalid = true;

×

108

  return tSimpleHashRemove(pBuff, pKey, keyLen);

×

111

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

×

112

  size_t        keyLen = pFileState->keyLen;

×

113

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

×

114

  if (ppPos) {

×

115

    if ((*ppPos) == pPos) {

×

116

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

×

117

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

×

120

×

122

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

×

124

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

×

126

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

×

127

  return streamStateDel_rocksdb(pFileState->pFileStore, pKey);

×

130

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

×

131

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

×

134

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

×

135

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

×

136

  if (pStateKey == NULL) {

×

137

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

×

138

    return NULL;

×

140

  SWinKey* pWinKey = pPos->pKey;

×

141

  pStateKey->key = *pWinKey;

×

142

  pStateKey->opNum = num;

×

143

  return pStateKey;

×

146

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

×

147

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

×

148

  if (pStateKey == NULL) {

×

149

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

×

150

    return NULL;

×

152

  SWinKey* pWinKey = pPos->pKey;

×

153

  *pStateKey = *pWinKey;

×

154

  return pStateKey;

×

157

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

×

158

  return streamStateSessionDel_rocksdb(pFileState->pFileStore, pKey);

×

161

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

×

162

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

×

165

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

×

166

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

×

167

  if (pStateKey == NULL) {

×

168

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

×

169

    return NULL;

×

171

  SSessionKey* pWinKey = pPos->pKey;

×

172

  pStateKey->key = *pWinKey;

×

173

  pStateKey->opNum = num;

×

174

  return pStateKey;

×

177

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

×

179

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

×

180

  *pLen = sizeof(TSKEY);

×

181

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

×

182

  if ((*pVal) == NULL) {

×

183

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

×

184

    return terrno;

×

186

  void*   buff = *pVal;

×

187

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

×

188

  return TSDB_CODE_SUCCESS;

×

191

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

×

194

  int32_t code = TSDB_CODE_SUCCESS;

×

195

  int32_t lino = 0;

×

196

  if (memSize <= 0) {

×

197

    memSize = DEFAULT_MAX_STREAM_BUFFER_SIZE;

×

199

  if (rowSize == 0) {

×

200

    code = TSDB_CODE_INVALID_PARA;

×

201

    QUERY_CHECK_CODE(code, lino, _end);

×

204

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

×

205

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

×

207

  rowSize += selectRowSize;

×

208

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

×

209

  pFileState->usedBuffs = tdListNew(POINTER_BYTES);

×

210

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

×

212

  pFileState->freeBuffs = tdListNew(POINTER_BYTES);

×

213

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

×

215

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

×

216

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

×

217

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

×

218

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

×

219

    pFileState->stateBuffCleanupFn = stateHashBuffCleanupFn;

×

220

    pFileState->stateBuffRemoveFn = stateHashBuffRemoveFn;

×

221

    pFileState->stateBuffRemoveByPosFn = stateHashBuffRemoveByPosFn;

×

222

    pFileState->stateBuffCreateStateKeyFn = intervalCreateStateKey;

×

224

    pFileState->stateFileRemoveFn = intervalFileRemoveFn;

×

225

    pFileState->stateFileGetFn = intervalFileGetFn;

×

226

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

×

227

    pFileState->stateFunctionGetFn = addRowBuffIfNotExist;

×

228

  } else if (type == STREAM_STATE_BUFF_SORT) {

×

229

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

×

230

    pFileState->stateBuffCleanupFn = sessionWinStateCleanup;

×

231

    pFileState->stateBuffRemoveFn = deleteSessionWinStateBuffFn;

×

232

    pFileState->stateBuffRemoveByPosFn = deleteSessionWinStateBuffByPosFn;

×

233

    pFileState->stateBuffCreateStateKeyFn = sessionCreateStateKey;

×

235

    pFileState->stateFileRemoveFn = sessionFileRemoveFn;

×

236

    pFileState->stateFileGetFn = sessionFileGetFn;

×

237

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

×

238

    pFileState->stateFunctionGetFn = getSessionRowBuff;

×

239

  } else if (type == STREAM_STATE_BUFF_HASH_SORT) {

×

240

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

×

241

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

×

242

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

×

243

    pFileState->stateBuffCleanupFn = stateHashBuffCleanupFn;

×

244

    pFileState->stateBuffRemoveFn = stateHashBuffRemoveFn;

×

245

    pFileState->stateBuffRemoveByPosFn = stateHashBuffRemoveByPosFn;

×

246

    pFileState->stateBuffCreateStateKeyFn = defaultCreateStateKey;

×

248

    pFileState->stateFileRemoveFn = hashSortFileRemoveFn;

×

249

    pFileState->stateFileGetFn = hashSortFileGetFn;

×

250

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

×

251

    pFileState->stateFunctionGetFn = NULL;

×

254

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

×

255

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

×

256

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

×

257

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

×

259

  if (type == STREAM_STATE_BUFF_HASH_SEARCH) {

×

260

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

×

261

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

×

264

  pFileState->keyLen = keySize;

×

265

  pFileState->rowSize = rowSize;

×

266

  pFileState->selectivityRowSize = selectRowSize;

×

267

  pFileState->preCheckPointVersion = 0;

×

268

  pFileState->checkPointVersion = 1;

×

269

  pFileState->pFileStore = pFile;

×

270

  pFileState->getTs = fp;

×

271

  pFileState->curRowCount = 0;

×

272

  pFileState->deleteMark = delMark;

×

273

  pFileState->flushMark = INT64_MIN;

×

274

  pFileState->maxTs = INT64_MIN;

×

275

  pFileState->id = taosStrdup(taskId);

×

276

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

×

278

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

×

279

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

×

281

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

×

282

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

×

285

  pFileState->hasFillCatch = true;

×

287

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

×

288

    code = recoverSnapshot(pFileState, checkpointId);

×

289

  } else if (type == STREAM_STATE_BUFF_SORT) {

×

290

    code = recoverSession(pFileState, checkpointId);

×

291

  } else if (type == STREAM_STATE_BUFF_HASH_SORT) {

×

292

    code = recoverFillSnapshot(pFileState, checkpointId);

×

294

  QUERY_CHECK_CODE(code, lino, _end);

×

296

  void*   valBuf = NULL;

×

297

  int32_t len = 0;

×

298

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

×

299

  if (tmpRes == TSDB_CODE_SUCCESS) {

×

300

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

×

301

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

×

302

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

×

304

  taosMemoryFreeClear(valBuf);

×

305

  (*ppFileState) = pFileState;

×

307

_end:

×

308

  if (code != TSDB_CODE_SUCCESS) {

×

309

    streamFileStateDestroy(pFileState);

×

310

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

×

312

  return code;

×

315

void destroyRowBuffPos(SRowBuffPos* pPos) {

×

316

  taosMemoryFreeClear(pPos->pKey);

×

317

  taosMemoryFreeClear(pPos->pRowBuff);

×

318

  taosMemoryFree(pPos);

×

319

×

321

void destroyRowBuffPosPtr(void* ptr) {

×

322

  if (!ptr) {

×

323

    return;

×

325

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

×

326

  if (!pPos->beUsed) {

×

327

    destroyRowBuffPos(pPos);

×

331

void destroyRowBuffAllPosPtr(void* ptr) {

×

332

  if (!ptr) {

×

333

    return;

×

335

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

×

336

  destroyRowBuffPos(pPos);

×

339

void destroyRowBuff(void* ptr) {

×

340

  if (!ptr) {

×

341

    return;

×

343

  taosMemoryFree(*(void**)ptr);

×

346

void streamFileStateDestroy(SStreamFileState* pFileState) {

×

347

  if (!pFileState) {

×

348

    return;

×

351

  taosMemoryFree(pFileState->id);

×

352

  taosMemoryFree(pFileState->cfName);

×

353

  tdListFreeP(pFileState->usedBuffs, destroyRowBuffAllPosPtr);

×

354

  tdListFreeP(pFileState->freeBuffs, destroyRowBuff);

×

355

  pFileState->stateBuffCleanupFn(pFileState->rowStateBuff);

×

356

  sessionWinStateCleanup(pFileState->searchBuff);

×

357

  tSimpleHashCleanup(pFileState->pGroupIdMap);

×

358

  tSimpleHashCleanup(pFileState->pRecFlagMap);

×

359

  taosMemoryFree(pFileState);

×

362

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

×

363

  int32_t code = TSDB_CODE_SUCCESS;

×

364

  int32_t lino = 0;

×

365

  if (pPos->pRowBuff) {

×

366

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

×

367

    QUERY_CHECK_CODE(code, lino, _end);

×

368

    pPos->pRowBuff = NULL;

×

371

_end:

×

372

  if (code != TSDB_CODE_SUCCESS) {

×

373

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

×

375

  return code;

×

378

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

×

379

  int32_t   code = TSDB_CODE_SUCCESS;

×

380

  int32_t   lino = 0;

×

381

  SListIter iter = {0};

×

382

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

×

384

  SListNode* pNode = NULL;

×

385

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

×

386

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

×

387

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

×

388

      code = putFreeBuff(pFileState, pPos);

×

389

      QUERY_CHECK_CODE(code, lino, _end);

×

391

      if (!all) {

×

392

        pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

×

394

      destroyRowBuffPos(pPos);

×

395

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

×

396

      taosMemoryFreeClear(tmp);

×

400

_end:

×

401

  if (code != TSDB_CODE_SUCCESS) {

×

402

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

×

404

×

406

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

×

407

  int32_t   code = TSDB_CODE_SUCCESS;

×

408

  int32_t   lino = 0;

×

409

  uint64_t  i = 0;

×

410

  SListIter iter = {0};

×

411

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

×

413

  SListNode* pNode = NULL;

×

414

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

×

415

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

×

416

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

×

417

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

×

418

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

×

419

          continue;

×

421

        code = tdListAppend(pFlushList, &pPos);

×

422

        QUERY_CHECK_CODE(code, lino, _end);

×

424

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

×

425

        pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

×

426

        if (pPos->beUsed == false) {

×

427

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

×

428

          taosMemoryFreeClear(tmp);

×

430

        if (pPos->pRowBuff) {

×

431

          i++;

×

436

  qDebug("clear flushed row buff. %d rows to disk. is all:%d", listNEles(pFlushList), all);

×

438

_end:

×

439

  if (code != TSDB_CODE_SUCCESS) {

×

440

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

×

442

  return code;

×

445

void streamFileStateClear(SStreamFileState* pFileState) {

×

446

  pFileState->flushMark = INT64_MIN;

×

447

  pFileState->maxTs = INT64_MIN;

×

448

  tSimpleHashClear(pFileState->rowStateBuff);

×

449

  clearExpiredRowBuff(pFileState, 0, true);

×

450

×

452

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

×

454

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

×

456

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

×

457

  int32_t   code = TSDB_CODE_SUCCESS;

×

458

  int32_t   lino = 0;

×

459

  uint64_t  i = 0;

×

460

  SListIter iter = {0};

×

461

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

×

463

  SListNode* pNode = NULL;

×

464

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

×

465

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

×

466

    if (pPos->beUsed == used) {

×

467

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

×

468

        continue;

×

470

      code = tdListAppend(pFlushList, &pPos);

×

471

      QUERY_CHECK_CODE(code, lino, _end);

×

473

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

×

474

      pFileState->stateBuffRemoveByPosFn(pFileState, pPos);

×

475

      if (pPos->beUsed == false) {

×

476

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

×

477

        taosMemoryFreeClear(tmp);

×

479

      if (pPos->pRowBuff) {

×

480

        i++;

×

485

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

×

487

_end:

×

488

  if (code != TSDB_CODE_SUCCESS) {

×

489

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

×

491

  return code;

×

494

int32_t flushRowBuff(SStreamFileState* pFileState) {

×

495

  int32_t          code = TSDB_CODE_SUCCESS;

×

496

  int32_t          lino = 0;

×

497

  SStreamSnapshot* pFlushList = tdListNew(POINTER_BYTES);

×

498

  if (!pFlushList) {

×

499

    code = TSDB_CODE_OUT_OF_MEMORY;

×

500

    QUERY_CHECK_CODE(code, lino, _end);

×

503

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

×

504

  num = TMAX(num, FLUSH_NUM);

×

505

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

×

506

  QUERY_CHECK_CODE(code, lino, _end);

×

508

  if (isListEmpty(pFlushList)) {

×

509

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

×

510

    QUERY_CHECK_CODE(code, lino, _end);

×

512

    if (isListEmpty(pFlushList)) {

×

513

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

×

514

      QUERY_CHECK_CODE(code, lino, _end);

×

518

  if (pFileState->searchBuff) {

×

519

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

×

520

    QUERY_CHECK_CODE(code, lino, _end);

×

523

  flushSnapshot(pFileState, pFlushList, false);

×

525

  SListIter fIter = {0};

×

526

  tdListInitIter(pFlushList, &fIter, TD_LIST_FORWARD);

×

527

  SListNode* pNode = NULL;

×

528

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

×

529

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

×

530

    code = putFreeBuff(pFileState, pPos);

×

531

    QUERY_CHECK_CODE(code, lino, _end);

×

534

  tdListFreeP(pFlushList, destroyRowBuffPosPtr);

×

536

_end:

×

537

  if (code != TSDB_CODE_SUCCESS) {

×

538

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

×

540

  return code;

×

543

int32_t clearRowBuff(SStreamFileState* pFileState) {

×

544

  if (pFileState->deleteMark != INT64_MAX) {

×

545

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

×

548

    int32_t code = flushRowBuff(pFileState);

×

549

    if (code != TSDB_CODE_SUCCESS) {

×

550

      return code;

×

552

  } while (isListEmpty(pFileState->freeBuffs) && pFileState->curRowCount == pFileState->maxRowCount);

×

553

  return TSDB_CODE_SUCCESS;

×

556

int32_t clearFlushedRowBuffByFlag(SStreamFileState* pFileState, uint64_t max) {

×

557

  int32_t   code = TSDB_CODE_SUCCESS;

×

558

  int32_t   lino = 0;

×

559

  uint64_t  i = 0;

×

560

  SListIter iter = {0};

×

561

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

×

563

  SListNode* pNode = NULL;

×

564

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

×

565

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

×

566

    if (pPos->invalid) {

×

567

      if (!pPos->beUsed) {

×

568

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

×

569

        taosMemoryFreeClear(tmp);

×

570

        if (pPos->pRowBuff) {

×

571

          i++;

×

573

        code = putFreeBuff(pFileState, pPos);

×

574

        QUERY_CHECK_CODE(code, lino, _end);

×

575

        destroyRowBuffPos(pPos);

×

580

_end:

×

581

  if (code != TSDB_CODE_SUCCESS) {

×

582

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

×

584

  return code;

×

587

int32_t clearRowBuffNonFlush(SStreamFileState* pFileState) {

×

588

  int32_t code = TSDB_CODE_SUCCESS;

×

589

  int32_t lino = 0;

×

591

  if (pFileState->deleteMark != INT64_MAX) {

×

592

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

×

595

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

×

596

  num = TMAX(num, FLUSH_NUM);

×

597

  code = clearFlushedRowBuffByFlag(pFileState, num);

×

598

  QUERY_CHECK_CODE(code, lino, _end);

×

600

_end:

×

601

  if (code != TSDB_CODE_SUCCESS) {

×

602

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

×

604

  return code;

×

607

void* getFreeBuff(SStreamFileState* pFileState) {

×

608

  SList*     lists = pFileState->freeBuffs;

×

609

  int32_t    buffSize = pFileState->rowSize;

×

610

  SListNode* pNode = tdListPopHead(lists);

×

611

  if (!pNode) {

×

612

    return NULL;

×

614

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

×

615

  memset(ptr, 0, buffSize);

×

616

  taosMemoryFree(pNode);

×

617

  return ptr;

×

620

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

×

621

  if (pPos->pRowBuff) {

×

622

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

×

624

×

626

SRowBuffPos* getNewRowPos(SStreamFileState* pFileState) {

×

627

  int32_t      code = TSDB_CODE_SUCCESS;

×

628

  int32_t      lino = 0;

×

629

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

×

630

  if (!pPos) {

×

631

    code = terrno;

×

632

    QUERY_CHECK_CODE(code, lino, _error);

×

635

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

×

636

  if (!pPos->pKey) {

×

637

    code = terrno;

×

638

    QUERY_CHECK_CODE(code, lino, _error);

×

641

  void* pBuff = getFreeBuff(pFileState);

×

642

  if (pBuff) {

×

643

    pPos->pRowBuff = pBuff;

×

644

    goto _end;

×

647

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

×

648

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

×

649

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

×

650

    pPos->pRowBuff = pBuff;

×

651

    pFileState->curRowCount++;

×

652

    goto _end;

×

655

  code = clearRowBuff(pFileState);

×

656

  QUERY_CHECK_CODE(code, lino, _error);

×

658

  pPos->pRowBuff = getFreeBuff(pFileState);

×

660

_end:

×

661

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

×

662

  QUERY_CHECK_CODE(code, lino, _error);

×

664

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

×

665

_error:

×

666

  if (code != TSDB_CODE_SUCCESS) {

×

667

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

×

668

    return NULL;

×

671

  return pPos;

×

674

SRowBuffPos* getNewRowPosForWrite(SStreamFileState* pFileState) {

×

675

  int32_t      code = TSDB_CODE_SUCCESS;

×

676

  int32_t      lino = 0;

×

677

  SRowBuffPos* newPos = getNewRowPos(pFileState);

×

678

  if (!newPos) {

×

679

    code = TSDB_CODE_OUT_OF_MEMORY;

×

680

    QUERY_CHECK_CODE(code, lino, _error);

×

682

  newPos->beUsed = true;

×

683

  newPos->beFlushed = false;

×

684

  newPos->needFree = false;

×

685

  newPos->beUpdated = true;

×

686

  newPos->invalid = false;

×

687

  return newPos;

×

689

_error:

×

690

  if (code != TSDB_CODE_SUCCESS) {

×

691

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

×

693

  return NULL;

×

696

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

×

698

  int32_t code = TSDB_CODE_SUCCESS;

×

699

  int32_t lino = 0;

×

700

  (*pWinCode) = TSDB_CODE_SUCCESS;

×

701

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

×

702

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

×

703

  if (pos) {

×

704

    if (pVal != NULL) {

×

705

      *pVLen = pFileState->rowSize;

×

706

      *pVal = *pos;

×

707

      (*pos)->beUsed = true;

×

708

      (*pos)->beFlushed = false;

×

710

    goto _end;

×

712

  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

713

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

×

714

    code = TSDB_CODE_OUT_OF_MEMORY;

×

715

    QUERY_CHECK_CODE(code, lino, _end);

×

718

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

×

719

  (*pWinCode) = TSDB_CODE_FAILED;

×

721

  TSKEY ts = pFileState->getTs(pKey);

×

722

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

×

723

    int32_t len = 0;

×

724

    void*   p = NULL;

×

725

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

×

726

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

×

727

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

728

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

×

730

    taosMemoryFree(p);

×

733

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

×

734

  QUERY_CHECK_CODE(code, lino, _end);

×

736

  if (pVal) {

×

737

    *pVLen = pFileState->rowSize;

×

738

    *pVal = pNewPos;

×

741

_end:

×

742

  if (code != TSDB_CODE_SUCCESS) {

×

743

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

×

745

  return code;

×

748

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

×

749

  int32_t code = TSDB_CODE_SUCCESS;

×

750

  int32_t lino = 0;

×

752

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

×

754

  SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

755

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

×

756

    code = TSDB_CODE_OUT_OF_MEMORY;

×

757

    QUERY_CHECK_CODE(code, lino, _end);

×

760

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

×

762

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

×

763

  QUERY_CHECK_CODE(code, lino, _end);

×

765

  if (pVal) {

×

766

    *pVLen = pFileState->rowSize;

×

767

    *pVal = pNewPos;

×

770

_end:

×

771

  if (code != TSDB_CODE_SUCCESS) {

×

772

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

×

774

  return code;

×

777

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

×

778

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

×

779

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

×

780

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

×

781

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

×

782

  if (pFileState->searchBuff != NULL) {

×

783

    deleteHashSortRowBuff(pFileState, pKey);

×

785

×

787

void deleteRowBuffByGroupId(SStreamFileState* pFileState, uint64_t groupId) {

×

788

  SSHashObj* pRowMap = pFileState->rowStateBuff;

×

789

  void*   pIte = NULL;

×

790

  int32_t iter = 0;

×

791

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

×

792

    size_t keyLen = 0;

×

793

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

×

794

    if (pKey->groupId == groupId) {

×

795

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

×

796

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

×

800

  while (1) {

×

801

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

×

802

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

×

803

    SWinKey delKey = {.groupId = groupId};

×

804

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

×

805

    if (code != TSDB_CODE_SUCCESS) {

×

806

      break;

×

808

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

×

809

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

×

811

×

813

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

×

814

  int32_t code = TSDB_CODE_SUCCESS;

×

815

  int32_t lino = 0;

×

816

  int32_t len = 0;

×

817

  void*   pBuff = NULL;

×

818

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

×

819

  QUERY_CHECK_CODE(code, lino, _end);

×

820

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

×

821

  taosMemoryFree(pBuff);

×

823

_end:

×

824

  if (code != TSDB_CODE_SUCCESS) {

×

825

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

×

827

  return code;

×

830

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

×

831

  int32_t code = TSDB_CODE_SUCCESS;

×

832

  int32_t lino = 0;

×

833

  pPos->pRowBuff = getFreeBuff(pFileState);

×

834

  if (!pPos->pRowBuff) {

×

835

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

×

836

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

×

837

      if (!pPos->pRowBuff) {

×

838

        code = terrno;

×

839

        QUERY_CHECK_CODE(code, lino, _end);

×

841

      pFileState->curRowCount++;

×

843

      code = clearRowBuff(pFileState);

×

844

      QUERY_CHECK_CODE(code, lino, _end);

×

845

      pPos->pRowBuff = getFreeBuff(pFileState);

×

847

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

×

850

  code = recoverSessionRowBuff(pFileState, pPos);

×

851

  QUERY_CHECK_CODE(code, lino, _end);

×

853

_end:

×

854

  if (code != TSDB_CODE_SUCCESS) {

×

855

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

×

857

  return code;

×

860

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

×

861

  int32_t code = TSDB_CODE_SUCCESS;

×

862

  int32_t lino = 0;

×

863

  if (pPos->pRowBuff) {

×

864

    if (pPos->needFree) {

×

865

      code = recoverSessionRowBuff(pFileState, pPos);

×

866

      QUERY_CHECK_CODE(code, lino, _end);

×

868

    (*pVal) = pPos->pRowBuff;

×

869

    goto _end;

×

872

  code = recoverStateRowBuff(pFileState, pPos);

×

873

  QUERY_CHECK_CODE(code, lino, _end);

×

875

  (*pVal) = pPos->pRowBuff;

×

881

_end:

×

882

  if (code != TSDB_CODE_SUCCESS) {

×

883

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

×

885

  return code;

×

888

bool hasRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, bool hasLimit, bool* pIsLast) {

×

889

  bool res = false;

×

890

  if (pIsLast != NULL) {

×

891

    (*pIsLast) = false;

×

894

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

×

895

  if (pos) {

×

896

    res = true;

×

898

  void* pSearchBuff = getSearchBuff(pFileState);

×

899

  if (pSearchBuff != NULL) {

×

900

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

×

901

    if (ppBuff != NULL) {

×

902

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

×

903

      if (pIsLast != NULL) {

×

904

        SWinKey* pLastKey = (SWinKey*)taosArrayGetLast(pWinStates);

×

905

        *pIsLast = (winKeyCmprImpl(pKey, pLastKey) == 0);

×

907

      if (hasLimit && taosArrayGetSize(pWinStates) <= MIN_NUM_OF_SORT_CACHE_WIN) {

×

908

        res = true;

×

910

      if (qDebugFlag & DEBUG_DEBUG) {

×

911

        if (taosArrayGetSize(pWinStates) > 0) {

×

912

          SWinKey* fistKey = (SWinKey*)taosArrayGet(pWinStates, 0);

×

913

          qDebug("===stream===check window state. buff min ts:%" PRId64 ",groupId:%" PRIu64 ".key ts:%" PRId64

×

919

      res = true;

×

922

  return res;

×

925

SStreamSnapshot* getSnapshot(SStreamFileState* pFileState) {

×

926

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

×

928

                     : pFileState->maxTs - pFileState->deleteMark;

×

929

  clearExpiredRowBuff(pFileState, mark, false);

×

930

  return pFileState->usedBuffs;

×

933

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

×

934

  int32_t   code = TSDB_CODE_SUCCESS;

×

935

  int32_t   lino = 0;

×

936

  SListIter iter = {0};

×

937

  tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD);

×

939

  int64_t    st = taosGetTimestampMs();

×

940

  SListNode* pNode = NULL;

×

942

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

×

944

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

×

945

  char*   buf = taosMemoryCalloc(1, len);

×

946

  if (!buf) {

×

947

    code = terrno;

×

948

    QUERY_CHECK_CODE(code, lino, _end);

×

951

  void* batch = streamStateCreateBatch();

×

952

  if (!batch) {

×

953

    code = TSDB_CODE_OUT_OF_MEMORY;

×

954

    QUERY_CHECK_CODE(code, lino, _end);

×

957

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

×

958

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

×

959

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

×

960

      continue;

×

962

    pPos->beFlushed = true;

×

963

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

×

965

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

×

966

    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {

×

967

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

×

968

      streamStateClearBatch(batch);

×

969

      QUERY_CHECK_CODE(code, lino, _end);

×

972

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

×

973

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

×

975

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

×

977

    taosMemoryFreeClear(pSKey);

×

978

    QUERY_CHECK_CODE(code, lino, _end);

×

980

    memset(buf, 0, len);

×

982

  taosMemoryFreeClear(buf);

×

984

  int32_t numOfElems = streamStateGetBatchSize(batch);

×

985

  if (numOfElems > 0) {

×

986

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

×

987

    QUERY_CHECK_CODE(code, lino, _end);

×

989

    goto _end;

×

992

  streamStateClearBatch(batch);

×

994

  clearSearchBuff(pFileState);

×

996

  int64_t elapsed = taosGetTimestampMs() - st;

×

997

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

×

1000

  if (flushState) {

×

1001

    void*   valBuf = NULL;

×

1002

    int32_t len = 0;

×

1003

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

×

1004

    QUERY_CHECK_CODE(code, lino, _end);

×

1006

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

×

1007

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

×

1008

    taosMemoryFree(valBuf);

×

1009

    QUERY_CHECK_CODE(code, lino, _end);

×

1011

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

×

1012

    QUERY_CHECK_CODE(code, lino, _end);

×

1015

_end:

×

1016

  if (code != TSDB_CODE_SUCCESS) {

×

1017

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

×

1019

  taosMemoryFree(buf);

×

1020

  streamStateDestroyBatch(batch);

×

1021

×

1023

int32_t forceRemoveCheckpoint(SStreamFileState* pFileState, int64_t checkpointId) {

×

1024

  char keyBuf[128] = {0};

×

1025

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

×

1026

  return streamDefaultDel_rocksdb(pFileState->pFileStore, keyBuf);

×

1029

int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {

×

1030

  int32_t code = TSDB_CODE_SUCCESS;

×

1031

  int64_t maxCheckPointId = 0;

×

1033

    char    buf[128] = {0};

×

1034

    void*   val = NULL;

×

1035

    int32_t len = 0;

×

1036

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

×

1037

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

×

1038

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

×

1039

      return TSDB_CODE_FAILED;

×

1041

    memcpy(buf, val, len);

×

1042

    buf[len] = 0;

×

1043

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

×

1044

    taosMemoryFree(val);

×

1046

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

×

1047

    char    buf[128] = {0};

×

1048

    void*   val = 0;

×

1049

    int32_t len = 0;

×

1050

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

×

1051

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

×

1052

    if (code != 0) {

×

1053

      return TSDB_CODE_FAILED;

×

1055

    memcpy(buf, val, len);

×

1056

    buf[len] = 0;

×

1057

    taosMemoryFree(val);

×

1060

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

×

1061

    if (ts < mark) {

×

1063

      int32_t tmpRes = forceRemoveCheckpoint(pFileState, i);

×

1064

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

×

1065

      break;

×

1068

  return code;

×

1071

int32_t recoverSession(SStreamFileState* pFileState, int64_t ckId) {

×

1072

  int32_t code = TSDB_CODE_SUCCESS;

×

1073

  int32_t lino = 0;

×

1074

  int32_t winRes = TSDB_CODE_SUCCESS;

×

1075

  if (pFileState->maxTs != INT64_MIN) {

×

1076

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

×

1078

                       : pFileState->maxTs - pFileState->deleteMark;

×

1079

    int32_t tmpRes = deleteExpiredCheckPoint(pFileState, mark);

×

1080

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

×

1083

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

×

1084

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

×

1085

  while (winRes == TSDB_CODE_SUCCESS) {

×

1086

    if (pFileState->curRowCount >= recoverNum) {

×

1087

      break;

×

1090

    void*       pVal = NULL;

×

1091

    int32_t     vlen = 0;

×

1092

    SSessionKey key = {0};

×

1093

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

×

1094

    if (winRes != TSDB_CODE_SUCCESS) {

×

1095

      break;

×

1098

    if (vlen != pFileState->rowSize) {

×

1099

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1100

      qError("[InternalERR] read key:[skey:%"PRId64 ",ekey:%"PRId64 ",groupId:%"PRIu64 "],vlen:%d, rowSize:%d", key.win.skey, key.win.ekey, key.groupId, vlen, pFileState->rowSize);

×

1101

      QUERY_CHECK_CODE(code, lino, _end);

×

1104

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

×

1105

    pPos->beUsed = false;

×

1106

    winRes = putSessionWinResultBuff(pFileState, pPos);

×

1107

    if (winRes != TSDB_CODE_SUCCESS) {

×

1108

      break;

×

1111

    winRes = streamStateSessionCurPrev_rocksdb(pCur);

×

1114

_end:

×

1115

  if (code != TSDB_CODE_SUCCESS) {

×

1116

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

×

1118

  streamStateFreeCur(pCur);

×

1119

  return code;

×

1122

int32_t recoverSnapshot(SStreamFileState* pFileState, int64_t ckId) {

×

1123

  int32_t code = TSDB_CODE_SUCCESS;

×

1124

  int32_t lino = 0;

×

1125

  int32_t winCode = TSDB_CODE_SUCCESS;

×

1126

  if (pFileState->maxTs != INT64_MIN) {

×

1127

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

×

1129

                       : pFileState->maxTs - pFileState->deleteMark;

×

1130

    int32_t tmpRes = deleteExpiredCheckPoint(pFileState, mark);

×

1131

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

×

1134

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

×

1135

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

×

1136

  while (winCode == TSDB_CODE_SUCCESS) {

×

1137

    if (pFileState->curRowCount >= recoverNum) {

×

1138

      break;

×

1141

    void*        pVal = NULL;

×

1142

    int32_t      vlen = 0;

×

1143

    SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1144

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

×

1145

      code = TSDB_CODE_OUT_OF_MEMORY;

×

1146

      QUERY_CHECK_CODE(code, lino, _end);

×

1150

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

×

1151

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

×

1152

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

×

1153

      destroyRowBuffPos(pNewPos);

×

1154

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

×

1155

      taosMemoryFreeClear(pNode);

×

1156

      taosMemoryFreeClear(pVal);

×

1157

      break;

×

1159

    if (vlen != pFileState->rowSize) {

×

1160

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

×

1161

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1162

      taosMemoryFreeClear(pVal);

×

1163

      QUERY_CHECK_CODE(code, lino, _end);

×

1165

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

×

1166

    taosMemoryFreeClear(pVal);

×

1167

    pNewPos->beFlushed = true;

×

1168

    pNewPos->beUsed = false;

×

1169

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

×

1170

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

×

1171

    if (code != TSDB_CODE_SUCCESS) {

×

1172

      destroyRowBuffPos(pNewPos);

×

1173

      break;

×

1175

    streamStateCurPrev_rocksdb(pCur);

×

1178

_end:

×

1179

  if (code != TSDB_CODE_SUCCESS) {

×

1180

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

×

1182

  streamStateFreeCur(pCur);

×

1183

  return code;

×

1186

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

×

1188

void streamFileStateReloadInfo(SStreamFileState* pFileState, TSKEY ts) {

×

1189

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

×

1190

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

×

1191

×

1193

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

×

1194

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

×

1196

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

×

1198

bool isDeteled(SStreamFileState* pFileState, TSKEY ts) {

×

1199

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

×

1200

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

×

1203

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

×

1205

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

×

1207

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

×

1209

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

×

1210

  int32_t winCode = TSDB_CODE_SUCCESS;

×

1211

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

×

1214

int32_t recoverFillSnapshot(SStreamFileState* pFileState, int64_t ckId) {

×

1215

  int32_t code = TSDB_CODE_SUCCESS;

×

1216

  int32_t lino = 0;

×

1217

  if (pFileState->maxTs != INT64_MIN) {

×

1218

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

×

1220

                       : pFileState->maxTs - pFileState->deleteMark;

×

1221

    code = deleteExpiredCheckPoint(pFileState, mark);

×

1222

    QUERY_CHECK_CODE(code, lino, _end);

×

1225

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

×

1226

  if (pCur == NULL) {

×

1227

    return code;

×

1229

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

×

1230

  int32_t winRes = TSDB_CODE_SUCCESS;

×

1231

  while (winRes == TSDB_CODE_SUCCESS) {

×

1232

    if (pFileState->curRowCount >= recoverNum) {

×

1233

      break;

×

1236

    void*        pVal = NULL;

×

1237

    int32_t      vlen = 0;

×

1238

    SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1239

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

×

1240

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

×

1241

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

×

1242

      destroyRowBuffPos(pNewPos);

×

1243

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

×

1244

      taosMemoryFreeClear(pNode);

×

1245

      taosMemoryFreeClear(pVal);

×

1246

      break;

×

1249

    if (vlen != pFileState->rowSize) {

×

1250

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

×

1251

      destroyRowBuffPos(pNewPos);

×

1252

      code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;

×

1253

      taosMemoryFreeClear(pVal);

×

1254

      QUERY_CHECK_CODE(code, lino, _end);

×

1257

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

×

1258

    taosMemoryFreeClear(pVal);

×

1259

    pNewPos->beFlushed = true;

×

1260

    pNewPos->beUsed = false;

×

1261

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

×

1262

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

×

1263

    if (winRes != TSDB_CODE_SUCCESS) {

×

1264

      destroyRowBuffPos(pNewPos);

×

1265

      break;

×

1267

    streamStateCurPrev_rocksdb(pCur);

×

1269

  streamStateFreeCur(pCur);

×

1271

_end:

×

1272

  if (code != TSDB_CODE_SUCCESS) {

×

1273

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

×

1275

  return code;

×

1278

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

×

1280

  int32_t code = TSDB_CODE_SUCCESS;

×

1281

  int32_t lino = 0;

×

1282

  (*pWinCode) = TSDB_CODE_FAILED;

×

1283

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

×

1284

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

×

1285

  if (ppPos) {

×

1286

    *pVLen = pFileState->rowSize;

×

1287

    *pVal = *ppPos;

×

1288

    (*ppPos)->beUsed = true;

×

1289

    (*ppPos)->beFlushed = false;

×

1290

    (*pWinCode) = TSDB_CODE_SUCCESS;

×

1291

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

×

1292

      code = recoverStateRowBuff(pFileState, *ppPos);

×

1293

      QUERY_CHECK_CODE(code, lino, _end);

×

1295

    goto _end;

×

1297

  TSKEY ts = pFileState->getTs(pKey);

×

1298

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

×

1299

    int32_t len = 0;

×

1300

    void*   p = NULL;

×

1301

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

×

1302

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

×

1303

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

1304

      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1305

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

×

1306

        code = TSDB_CODE_OUT_OF_MEMORY;

×

1307

        QUERY_CHECK_CODE(code, lino, _end);

×

1310

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

×

1311

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

×

1312

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

×

1313

      QUERY_CHECK_CODE(code, lino, _end);

×

1315

      if (pVal) {

×

1316

        *pVLen = pFileState->rowSize;

×

1317

        *pVal = pNewPos;

×

1320

    taosMemoryFree(p);

×

1323

_end:

×

1324

  if (code != TSDB_CODE_SUCCESS) {

×

1325

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

×

1327

  return code;

×

1330

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

×

1331

  int32_t code = TSDB_CODE_SUCCESS;

×

1332

  int32_t lino = 0;

×

1333

  if (value != NULL) {

×

1334

    code = TSDB_CODE_INVALID_PARA;

×

1335

    QUERY_CHECK_CODE(code, lino, _end);

×

1338

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

×

1339

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

×

1340

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

×

1341

      QUERY_CHECK_CODE(code, lino, _end);

×

1343

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

×

1344

    QUERY_CHECK_CODE(code, lino, _end);

×

1347

_end:

×

1348

  if (code != TSDB_CODE_SUCCESS) {

×

1349

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

×

1351

  return code;

×

1354

void streamFileStateGroupCurNext(SStreamStateCur* pCur) {

×

1355

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

×

1356

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

×

1357

    streamStateCurNext(pFileState->pFileStore, pCur);

×

1358

    return;

×

1361

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

×

1362

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

×

1364

  SSHashObj* pHash = pFileState->pGroupIdMap;

×

1365

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

×

1366

  if (!pCur->pHashData) {

×

1367

    pCur->hashIter = -1;

×

1368

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

×

1369

    return;

×

1373

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

×

1374

  int32_t code = TSDB_CODE_SUCCESS;

×

1375

  if (pCur->pHashData) {

×

1376

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

×

1377

    return code;

×

1379

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

×

1382

SSHashObj* getGroupIdCache(SStreamFileState* pFileState) {

×

1383

  return pFileState->pGroupIdMap;

×

1386

void clearExpiredState(SStreamFileState* pFileState, int32_t numOfKeep, TSKEY minTs) {

×

1387

  int32_t    code = TSDB_CODE_SUCCESS;

×

1388

  int32_t    lino = 0;

×

1389

  SSHashObj* pSearchBuff = pFileState->searchBuff;

×

1390

  void*      pIte = NULL;

×

1391

  int32_t    iter = 0;

×

1392

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

×

1393

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

×

1394

    int32_t arraySize = TARRAY_SIZE(pWinStates);

×

1395

    if (minTs != INT64_MAX && arraySize > numOfKeep) {

×

1396

      SWinKey key = {.ts = minTs};

×

1397

      key.groupId = *(uint64_t*)tSimpleHashGetKey(pIte, NULL);

×

1398

      int32_t index = binarySearch(pWinStates, arraySize, &key, fillStateKeyCompare);

×

1399

      numOfKeep = TMAX(arraySize - index, MIN_NUM_OF_SORT_CACHE_WIN);

×

1400

      qDebug("modify numOfKeep, numOfKeep:%d. %s at line %d", numOfKeep, __func__, __LINE__);

×

1403

    int32_t size = arraySize - numOfKeep;

×

1404

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

×

1405

      SWinKey* pKey = taosArrayGet(pWinStates, i);

×

1406

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

×

1407

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

×

1409

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

×

1410

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

×

1411

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

×

1414

      if (tSimpleHashGetSize(pFileState->pRecFlagMap) > 0) {

×

1415

        tSimpleHashRemove(pFileState->pRecFlagMap, pKey, sizeof(SWinKey));

×

1418

    taosArrayRemoveBatch(pWinStates, 0, size, NULL);

×

1420

  code = clearRowBuffNonFlush(pFileState);

×

1421

  QUERY_CHECK_CODE(code, lino, _end);

×

1423

_end:

×

1424

  if (code != TSDB_CODE_SUCCESS) {

×

1425

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

×

1427

×

1484

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

×

1486

  int32_t    code = TSDB_CODE_SUCCESS;

×

1487

  int32_t    lino = 0;

×

1488

  SArray*    pWinStates = NULL;

×

1489

  SSHashObj* pSearchBuff = getSearchBuff(pFileState);

×

1490

  void*      pState = getStateFileStore(pFileState);

×

1491

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

×

1492

  if (ppBuff) {

×

1493

    pWinStates = (SArray*)(*ppBuff);

×

1494

  } else if (needClearDiskBuff(pFileState)) {

×

1495

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

×

1496

    SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, pKey);

×

1497

    void*            tmpVal = NULL;

×

1498

    int32_t          len = 0;

×

1499

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

×

1500

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

1501

      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1502

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

×

1503

        code = TSDB_CODE_OUT_OF_MEMORY;

×

1504

        QUERY_CHECK_CODE(code, lino, _end);

×

1506

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

×

1507

      taosMemoryFreeClear(tmpVal);

×

1508

      *pVLen = getRowStateRowSize(pFileState);

×

1509

      (*ppVal) = pNewPos;

×

1511

    streamStateFreeCur(pCur);

×

1512

    return code;

×

1514

    (*pWinCode) = TSDB_CODE_FAILED;

×

1515

    return code;

×

1517

  int32_t size = taosArrayGetSize(pWinStates);

×

1518

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

×

1519

  if (index >= 0) {

×

1520

    SWinKey* pCurKey = taosArrayGet(pWinStates, index);

×

1521

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

×

1522

      index--;

×

1524

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

×

1528

  if (index == -1) {

×

1529

    SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, pKey);

×

1530

    void*            tmpVal = NULL;

×

1531

    int32_t          len = 0;

×

1532

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

×

1533

    if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

1534

      SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);

×

1535

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

×

1536

        code = TSDB_CODE_OUT_OF_MEMORY;

×

1537

        QUERY_CHECK_CODE(code, lino, _end);

×

1539

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

×

1540

      taosMemoryFreeClear(tmpVal);

×

1541

      *pVLen = getRowStateRowSize(pFileState);

×

1542

      (*ppVal) = pNewPos;

×

1544

    streamStateFreeCur(pCur);

×

1545

    return code;

×

1547

    SWinKey* pPrevKey = taosArrayGet(pWinStates, index);

×

1548

    *pResKey = *pPrevKey;

×

1549

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

×

1553

_end:

×

1554

  if (code != TSDB_CODE_SUCCESS) {

×

1555

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

×

1557

  return code;

×

1560

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

×

1561

  int32_t code = TSDB_CODE_SUCCESS;

×

1562

  int32_t lino = 0;

×

1563

  int32_t size = taosArrayGetSize(pWinStates);

×

1564

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

×

1565

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

×

1566

    if (index >= 0) {

×

1567

      SWinKey* pTmpKey = taosArrayGet(pWinStates, index);

×

1568

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

×

1569

        goto _end;

×

1572

    index++;

×

1573

    (*pIsEnd) = (index >= size);

×

1574

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

×

1575

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

×

1578

_end:

×

1579

  if (code != TSDB_CODE_SUCCESS) {

×

1580

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

×

1582

  return code;

×

1585

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

×

1586

  int32_t code = TSDB_CODE_SUCCESS;

×

1587

  int32_t lino = 0;

×

1588

  SArray* pWinStates = NULL;

×

1589

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

×

1590

  if (ppBuff) {

×

1591

    pWinStates = (SArray*)(*ppBuff);

×

1593

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

×

1594

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

×

1596

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

×

1597

    QUERY_CHECK_CODE(code, lino, _end);

×

1600

  (*ppResStates) = pWinStates;

×

1602

_end:

×

1603

  if (code != TSDB_CODE_SUCCESS) {

×

1604

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

×

1606

  return code;

×

1609

static void setValueBuff(TSKEY ts, char* pVal, int32_t len, char* pBuff, int32_t buffLen) {

×

1610

  SET_TSDATA_FLAG(pBuff, buffLen);

×

1611

  if (len == 0) {

×

1612

    *(TSKEY*)pBuff = ts;

×

1613

    return;

×

1615

  memset(pBuff, 0, buffLen - 1);

×

1616

  *(TSKEY*)pBuff = ts;

×

1617

  memcpy(pBuff + sizeof(TSKEY), pVal, len);

×

1620

int32_t getAndSetTsData(STableTsDataState* pTsDataState, uint64_t tableUid, TSKEY* pCurTs, void** ppCurPkVal,

×

1622

  int32_t code = TSDB_CODE_SUCCESS;

×

1623

  int32_t lino = 0;

×

1624

  bool    hasPk = (lastPkLen != 0);

×

1626

  TSKEY* pDataVal = tSimpleHashGet(pTsDataState->pTableTsDataMap, &tableUid, sizeof(uint64_t));

×

1627

  if (pDataVal != NULL) {

×

1628

    (*pWinCode) = TSDB_CODE_SUCCESS;

×

1629

    *pCurTs = *pDataVal;

×

1630

    if ((*pCurTs) < lastTs) {

×

1631

      setValueBuff(lastTs, pLastPkVal, lastPkLen, (char*)pDataVal, pTsDataState->pkValLen);

×

1633

      if (hasPk) {

×

1634

        (*ppCurPkVal) = POINTER_SHIFT(pDataVal, sizeof(TSKEY));

×

1635

        if ((*pCurTs) == lastTs && pTsDataState->comparePkColFn((*ppCurPkVal), pLastPkVal) < 0) {

×

1636

          setValueBuff(lastTs, pLastPkVal, lastPkLen, (char*)pDataVal, pTsDataState->pkValLen);

×

1641

    setValueBuff(lastTs, pLastPkVal, lastPkLen, pTsDataState->pPkValBuff, pTsDataState->pkValLen);

×

1642

    int32_t size = tSimpleHashGetSize(pTsDataState->pTableTsDataMap);

×

1643

    if (size < MAX_STATE_MAP_SIZE) {

×

1644

      (*pWinCode) = TSDB_CODE_FAILED;

×

1645

      code = tSimpleHashPut(pTsDataState->pTableTsDataMap, &tableUid, sizeof(uint64_t), pTsDataState->pPkValBuff,

×

1646

                            pTsDataState->pkValLen);

×

1647

      QUERY_CHECK_CODE(code, lino, _end);

×

1649

      (*pWinCode) = streamStateGetParTag_rocksdb(pTsDataState->pState, tableUid, &pTsDataState->pPkValBuff,

×

1651

      if ((*pWinCode) == TSDB_CODE_SUCCESS) {

×

1652

        *pCurTs = *(TSKEY*)pTsDataState->pPkValBuff;

×

1653

        if (hasPk) {

×

1654

          (*ppCurPkVal) = POINTER_SHIFT(pTsDataState->pPkValBuff, sizeof(TSKEY));

×

1658

      int32_t tmpCode = streamStatePutParTag_rocksdb(pTsDataState->pState, tableUid, pTsDataState->pPkValBuff,

×

1663

_end:

×

1664

  if (code != TSDB_CODE_SUCCESS) {

×

1665

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

×

1667

  return code;

×

1670

int32_t doTsDataCommit(STableTsDataState* pTsDataState) {

×

1671

  int32_t code = TSDB_CODE_SUCCESS;

×

1672

  int32_t lino = 0;

×

1673

  void*   batch = NULL;

×

1674

  char*   pTempBuf = NULL;

×

1676

  batch = streamStateCreateBatch();

×

1677

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

×

1678

  int           idx = streamStateGetCfIdx(pTsDataState->pState, "partag");

×

1679

  int32_t       len = (pTsDataState->pkValLen + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;

×

1680

  pTempBuf = taosMemoryCalloc(1, len);

×

1681

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

×

1683

  void*   pIte = NULL;

×

1684

  int32_t iter = 0;

×

1685

  while ((pIte = tSimpleHashIterate(pTsDataState->pTableTsDataMap, pIte, &iter)) != NULL) {

×

1686

    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {

×

1687

      code = streamStatePutBatch_rocksdb(pTsDataState->pState, batch);

×

1688

      streamStateClearBatch(batch);

×

1689

      QUERY_CHECK_CODE(code, lino, _end);

×

1692

    if (HAS_TSDATA_FLAG(pIte, pTsDataState->pkValLen)) {

×

1693

      void* pKey = tSimpleHashGetKey(pIte, NULL);

×

1694

      UNSET_TSDATA_FLAG(pIte, pTsDataState->pkValLen);

×

1695

      code = streamStatePutBatchOptimize(pTsDataState->pState, idx, batch, pKey, pIte, pTsDataState->pkValLen, 0,

×

1697

      QUERY_CHECK_CODE(code, lino, _end);

×

1698

      memset(pTempBuf, 0, len);

×

1699

      qDebug("flush ts data,table id:%" PRIu64 , *(uint64_t*)pKey);

×

1703

  int32_t numOfElems = streamStateGetBatchSize(batch);

×

1704

  if (numOfElems > 0) {

×

1705

    code = streamStatePutBatch_rocksdb(pTsDataState->pState, batch);

×

1706

    QUERY_CHECK_CODE(code, lino, _end);

×

1709

_end:

×

1710

  if (code != TSDB_CODE_SUCCESS) {

×

1711

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

×

1713

  taosMemoryFree(pTempBuf);

×

1714

  streamStateDestroyBatch(batch);

×

1715

  return code;

×

1718

int32_t doRangeDataCommit(STableTsDataState* pTsDataState) {

×

1719

  int32_t code = TSDB_CODE_SUCCESS;

×

1720

  int32_t lino = 0;

×

1721

  void*   batch = NULL;

×

1723

  batch = streamStateCreateBatch();

×

1724

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

×

1725

  int           idx = streamStateGetCfIdx(pTsDataState->pState, "sess");

×

1726

  int32_t       len = (pTsDataState->pkValLen + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;

×

1728

  int32_t size = taosArrayGetSize(pTsDataState->pScanRanges);

×

1729

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

×

1730

    SScanRange* pRange = taosArrayGet(pTsDataState->pScanRanges, i);

×

1731

    if (streamStateGetBatchSize(batch) >= BATCH_LIMIT) {

×

1732

      code = streamStatePutBatch_rocksdb(pTsDataState->pState, batch);

×

1733

      streamStateClearBatch(batch);

×

1734

      QUERY_CHECK_CODE(code, lino, _end);

×

1736

    SSessionKey key = {.win = pRange->win, .groupId = 0};

×

1737

    int32_t     uidSize = tSimpleHashGetSize(pRange->pUIds);

×

1738

    int32_t     gpIdSize = tSimpleHashGetSize(pRange->pGroupIds);

×

1739

    int32_t     size = uidSize + gpIdSize;

×

1740

    uint64_t*   pIdBuf = (uint64_t*)taosMemoryCalloc(1, size);

×

1741

    void*       pIte = NULL;

×

1742

    int32_t     iter = 0;

×

1743

    int32_t     i = 0;

×

1744

    while ((pIte = tSimpleHashIterate(pTsDataState->pTableTsDataMap, pIte, &iter)) != NULL) {

×

1745

      void* pTempKey = tSimpleHashGetKey(pIte, NULL);

×

1746

      pIdBuf[i] = *(uint64_t*)pTempKey;

×

1747

      i++;

×

1750

    code = streamStatePutBatchOptimize(pTsDataState->pState, idx, batch, &key, (void*)pIdBuf, size, 0,

×

1752

    QUERY_CHECK_CODE(code, lino, _end);

×

1755

  int32_t numOfElems = streamStateGetBatchSize(batch);

×

1756

  if (numOfElems > 0) {

×

1757

    code = streamStatePutBatch_rocksdb(pTsDataState->pState, batch);

×

1758

    QUERY_CHECK_CODE(code, lino, _end);

×

1761

_end:

×

1762

  if (code != TSDB_CODE_SUCCESS) {

×

1763

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

×

1765

  streamStateDestroyBatch(batch);

×

1766

  return code;

×

1769

int32_t initTsDataState(STableTsDataState** ppTsDataState, int8_t pkType, int32_t pkLen, void* pState, void* pOtherState) {

×

1770

  int32_t    code = TSDB_CODE_SUCCESS;

×

1771

  int32_t    lino = 0;

×

1773

  STableTsDataState* pTsDataState = taosMemoryCalloc(1, sizeof(STableTsDataState));

×

1774

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

×

1776

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT);

×

1777

  pTsDataState->pTableTsDataMap = tSimpleHashInit(DEFAULT_STATE_MAP_CAPACITY, hashFn);

×

1778

  QUERY_CHECK_NULL(pTsDataState->pTableTsDataMap, code, lino, _end, terrno);

×

1780

  pTsDataState->pkValLen = sizeof(TSKEY) + pkLen + sizeof(char);

×

1781

  pTsDataState->pPkValBuff = taosMemoryCalloc(1, pTsDataState->pkValLen);

×

1782

  QUERY_CHECK_NULL(pTsDataState->pPkValBuff, code, lino, _end, terrno);

×

1784

  if (pkLen != 0) {

×

1785

    pTsDataState->comparePkColFn = getKeyComparFunc(pkType, TSDB_ORDER_ASC);

×

1787

    pTsDataState->comparePkColFn = NULL;

×

1790

  pTsDataState->pScanRanges = taosArrayInit(64, sizeof(SScanRange));

×

1791

  QUERY_CHECK_NULL(pTsDataState->pScanRanges, code, lino, _end, terrno);

×

1793

  pTsDataState->pState = pState;

×

1794

  pTsDataState->recValueLen = sizeof(SRecDataInfo) + pkLen;

×

1795

  pTsDataState->pRecValueBuff = taosMemoryCalloc(1, pTsDataState->recValueLen);

×

1796

  QUERY_CHECK_NULL(pTsDataState->pRecValueBuff, code, lino, _end, terrno);

×

1798

  pTsDataState->curRecId = -1;

×

1800

  pTsDataState->pStreamTaskState = pOtherState;

×

1802

  pTsDataState->cfgIndex = streamStateGetCfIdx(pTsDataState->pState, "sess");

×

1803

  pTsDataState->pBatch = streamStateCreateBatch();

×

1804

  QUERY_CHECK_NULL(pTsDataState->pBatch, code, lino, _end, TSDB_CODE_FAILED);

×

1806

  pTsDataState->batchBufflen = (pTsDataState->recValueLen + sizeof(uint64_t) + sizeof(int32_t) + 64) * 2;

×

1807

  pTsDataState->pBatchBuff = taosMemoryCalloc(1, pTsDataState->batchBufflen);

×

1809

  (*ppTsDataState) = pTsDataState;

×

1811

_end:

×

1812

  if (code != TSDB_CODE_SUCCESS) {

×

1813

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

×

1815

  return code;

×

1818

static void destroyScanRange(SScanRange* pRange) {

×

1819

  pRange->win.skey = INT64_MIN;

×

1820

  pRange->win.ekey = INT64_MIN;

×

1821

  tSimpleHashCleanup(pRange->pUIds);

×

1822

  pRange->pUIds = NULL;

×

1823

  tSimpleHashCleanup(pRange->pGroupIds);

×

1824

  pRange->pGroupIds = NULL;

×

1825

×

1827

void destroyTsDataState(STableTsDataState* pTsDataState) {

×

1828

  SArray* pScanRanges = pTsDataState->pScanRanges;

×

1829

  int32_t size = taosArrayGetSize(pScanRanges);

×

1830

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

×

1831

    SScanRange* pRange = taosArrayGet(pScanRanges, i);

×

1832

    destroyScanRange(pRange);

×

1834

  taosArrayDestroy(pTsDataState->pScanRanges);

×

1835

  tSimpleHashCleanup(pTsDataState->pTableTsDataMap);

×

1836

  taosMemoryFreeClear(pTsDataState->pPkValBuff);

×

1837

  taosMemoryFreeClear(pTsDataState->pState);

×

1838

  taosMemoryFreeClear(pTsDataState->pRecValueBuff);

×

1839

  pTsDataState->pStreamTaskState = NULL;

×

1841

  streamStateClearBatch(pTsDataState->pBatch);

×

1842

  streamStateDestroyBatch(pTsDataState->pBatch);

×

1843

  pTsDataState->pBatch = NULL;

×

1844

  taosMemoryFreeClear(pTsDataState->pBatchBuff);

×

1846

  taosMemoryFreeClear(pTsDataState);

×

1847

×

1849

int32_t recoverTsData(STableTsDataState* pTsDataState) {

×

1850

  int32_t          code = TSDB_CODE_SUCCESS;

×

1851

  int32_t          lino = 0;

×

1852

  SStreamStateCur* pCur = createStateCursor(NULL);

×

1853

  streamStateParTagSeekKeyNext_rocksdb(pTsDataState->pState, INT64_MIN, pCur);

×

1854

  while (1) {

×

1855

    uint64_t tableUid = 0;

×

1856

    void*    pVal = NULL;

×

1857

    int32_t  len = 0;

×

1858

    int32_t  winCode = streamStateParTagGetKVByCur_rocksdb(pCur, &tableUid, &pVal, &len);

×

1859

    if (winCode != TSDB_CODE_SUCCESS) {

×

1860

      break;

×

1862

    if (pTsDataState->pkValLen != len) {

×

1863

      taosMemoryFree(pVal);

×

1864

      streamStateCurNext_rocksdb(pCur);

×

1865

      continue;

×

1867

    UNSET_TSDATA_FLAG(pVal, len);

×

1868

    code = tSimpleHashPut(pTsDataState->pTableTsDataMap, &tableUid, sizeof(uint64_t), pVal, len);

×

1869

    taosMemoryFree(pVal);

×

1870

    QUERY_CHECK_CODE(code, lino, _end);

×

1871

    streamStateCurNext_rocksdb(pCur);

×

1874

_end:

×

1875

  streamStateFreeCur(pCur);

×

1876

  if (code != TSDB_CODE_SUCCESS) {

×

1877

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

×

1879

  return code;

×

1882

SStreamStateCur* getLastStateCur(SStreamFileState* pFileState, getStateBuffFn fn) {

×

1883

  SStreamStateCur* pCur = createStateCursor(pFileState);

×

1884

  if (pCur == NULL) {

×

1885

    return NULL;

×

1887

  SSHashObj* pSearchBuff = fn(pFileState);

×

1888

  pCur->buffIndex = 0;

×

1889

  pCur->hashIter = 0;

×

1890

  pCur->pHashData = NULL;

×

1891

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

×

1892

  return pCur;

×

1895

void moveLastStateCurNext(SStreamStateCur* pCur, getStateBuffFn fn) {

×

1896

  SSHashObj* pSearchBuff = fn(pCur->pStreamFileState);

×

1897

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

×

1898

×

1900

int32_t getNLastStateKVByCur(SStreamStateCur* pCur, int32_t num, SArray* pRes) {

×

1901

  int32_t code = TSDB_CODE_SUCCESS;

×

1902

  int32_t lino = 0;

×

1903

  SArray*  pWinStates = NULL;

×

1904

  int32_t size = 0;

×

1907

    if (pCur->pHashData == NULL) {

×

1908

      return TSDB_CODE_FAILED;

×

1910

    pWinStates = *((void**)pCur->pHashData);

×

1911

    size = taosArrayGetSize(pWinStates);

×

1912

    if (size > 0) {

×

1913

      break;

×

1915

    moveLastStateCurNext(pCur, getSearchBuff);

×

1918

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

×

1920

  for ( ; i < size; i++) {

×

1921

    SWinKey* pKey = taosArrayGet(pWinStates, i);

×

1922

    int32_t  len = 0;

×

1923

    void*    pVal = NULL;

×

1924

    int32_t  winCode = TSDB_CODE_SUCCESS;

×

1925

    code = addRowBuffIfNotExist(pCur->pStreamFileState, (void*)pKey, sizeof(SWinKey), &pVal, &len, &winCode);

×

1926

    QUERY_CHECK_CODE(code, lino, _end);

×

1928

    if (winCode != TSDB_CODE_SUCCESS) {

×

1929

      qError("%s failed at line %d since window not exist. ts:%" PRId64 ",groupId:%" PRIu64, __func__, __LINE__,

×

1933

    void* pTempRes = taosArrayPush(pRes, &pVal);

×

1934

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

×

1937

_end:

×

1938

  if (code != TSDB_CODE_SUCCESS) {

×

1939

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

×

1941

  return code;

×

1944

int32_t reloadTsDataState(STableTsDataState* pTsDataState) {

×

1945

  int32_t code = TSDB_CODE_SUCCESS;

×

1946

  int32_t lino = 0;

×

1948

  STableTsDataState tmpState = *pTsDataState;

×

1949

  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT);

×

1950

  tmpState.pTableTsDataMap = tSimpleHashInit(DEFAULT_STATE_MAP_CAPACITY, hashFn);

×

1951

  QUERY_CHECK_NULL(tmpState.pTableTsDataMap, code, lino, _end, terrno);

×

1953

  code = recoverTsData(&tmpState);

×

1954

  QUERY_CHECK_CODE(code, lino, _end);

×

1956

  void*   pIte = NULL;

×

1957

  int32_t iter = 0;

×

1958

  while ((pIte = tSimpleHashIterate(pTsDataState->pTableTsDataMap, pIte, &iter)) != NULL) {

×

1959

    size_t keyLen = 0;

×

1960

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

×

1961

    code = tSimpleHashPut(tmpState.pTableTsDataMap, pKey, keyLen, pIte, pTsDataState->pkValLen);

×

1962

    QUERY_CHECK_CODE(code, lino, _end);

×

1964

  tSimpleHashCleanup(pTsDataState->pTableTsDataMap);

×

1965

  pTsDataState->pTableTsDataMap = tmpState.pTableTsDataMap;

×

1967

_end:

×

1968

  if (code != TSDB_CODE_SUCCESS) {

×

1969

    tSimpleHashCleanup(tmpState.pTableTsDataMap);

×

1970

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

×

1972

  return code;

×

1975

int32_t saveRecInfoToDisk(STableTsDataState* pTsDataState, SSessionKey* pKey, SRecDataInfo* pVal, int32_t vLen) {

×

1976

  int32_t code = TSDB_CODE_SUCCESS;

×

1977

  int32_t lino = 0;

×

1979

  SStateSessionKey stateKey = {.key = *pKey, .opNum = ((SStreamState*)pTsDataState->pState)->number};

×

1980

  code = streamStatePutBatchOptimize(pTsDataState->pState, pTsDataState->cfgIndex, pTsDataState->pBatch, &stateKey, pVal, vLen, 0,

×

1982

  QUERY_CHECK_CODE(code, lino, _end);

×

1984

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

×

1986

  if (streamStateGetBatchSize(pTsDataState->pBatch) >= BATCH_LIMIT) {

×

1987

    code = streamStatePutBatch_rocksdb(pTsDataState->pState, pTsDataState->pBatch);

×

1988

    streamStateClearBatch(pTsDataState->pBatch);

×

1989

    QUERY_CHECK_CODE(code, lino, _end);

×

1992

_end:

×

1993

  if (code != TSDB_CODE_SUCCESS) {

×

1994

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

×

1996

  return code;

×

1999

int32_t flushRemainRecInfoToDisk(STableTsDataState* pTsDataState) {

×

2000

  int32_t code = streamStatePutBatch_rocksdb(pTsDataState->pState, pTsDataState->pBatch);

×

2001

  streamStateClearBatch(pTsDataState->pBatch);

×

2002

  return code;

×

2005

int32_t recoverHashSortBuff(SStreamFileState* pFileState, SArray* pWinStates, uint64_t groupId) {

×

2006

  int32_t code = TSDB_CODE_SUCCESS;

×

2007

  int32_t lino = 0;

×

2009

  SWinKey          start = {.groupId = groupId, .ts = INT64_MAX};

×

2010

  void*            pState = getStateFileStore(pFileState);

×

2011

  SStreamStateCur* pCur = streamStateSeekKeyPrev_rocksdb(pState, &start);

×

2012

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

×

2013

    SWinKey tmpKey = {.groupId = groupId};

×

2014

    int32_t tmpRes = streamStateGetGroupKVByCur_rocksdb(pState, pCur, &tmpKey, NULL, 0);

×

2015

    if (tmpRes != TSDB_CODE_SUCCESS) {

×

2016

      break;

×

2018

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

×

2019

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

×

2020

    streamStateCurPrev_rocksdb(pCur);

×

2022

  taosArraySort(pWinStates, winKeyCmprImpl);

×

2023

  streamStateFreeCur(pCur);

×

2025

_end:

×

2026

  if (code != TSDB_CODE_SUCCESS) {

×

2027

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

×

2029

  return code;

×

2032

int32_t getRowStateAllPrevRow(SStreamFileState* pFileState, const SWinKey* pKey, SArray* pResArray, int32_t maxNum) {

×

2033

  int32_t    code = TSDB_CODE_SUCCESS;

×

2034

  int32_t    lino = 0;

×

2035

  SWinKey*   pPrevKey = NULL;

×

2036

  SSHashObj* pSearchBuff = getSearchBuff(pFileState);

×

2037

  void*      pState = getStateFileStore(pFileState);

×

2038

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

×

2039

  int32_t    num = 0;

×

2040

  if (ppBuff) {

×

2041

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

×

2042

    int32_t size = taosArrayGetSize(pWinStates);

×

2043

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

×

2044

    for (; index >= 0 && num <= maxNum; index--) {

×

2045

      pPrevKey = taosArrayGet(pWinStates, index);

×

2046

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

×

2047

        continue;

×

2049

      void*   pVal = NULL;

×

2050

      int32_t len = 0;

×

2051

      int32_t winCode = TSDB_CODE_SUCCESS;

×

2052

      code = addRowBuffIfNotExist(pFileState, (void*)pPrevKey, sizeof(SWinKey), &pVal, &len, &winCode);

×

2053

      QUERY_CHECK_CODE(code, lino, _end);

×

2054

      void* tempRes = taosArrayPush(pResArray, &pVal);

×

2055

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

×

2056

      num++;

×

2060

_end:

×

2061

  if (code != TSDB_CODE_SUCCESS) {

×

2062

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

×

2064

  return code;

×

2067

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

×

2068

  return tSimpleHashPut(pFileState->pRecFlagMap, pKey, keyLen, &mode, sizeof(int32_t));

×

2071

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

×

2072

  void* pVal = tSimpleHashGet(pFileState->pRecFlagMap, pKey, keyLen);

×

2073

  if (pVal == NULL) {

×

2074

    return TSDB_CODE_FAILED;

×

2076

  *pMode = *(int32_t*) pVal;

×

2077

  return TSDB_CODE_SUCCESS;

×

2080

void clearExpiredSessionState(SStreamFileState* pFileState, int32_t numOfKeep, TSKEY minTs, SSHashObj* pFlushGroup) {

×

2081

  int32_t    code = TSDB_CODE_SUCCESS;

×

2082

  int32_t    lino = 0;

×

2083

  SSHashObj* pSessionBuff = pFileState->rowStateBuff;

×

2084

  SStreamSnapshot* pFlushList = NULL;

×

2085

  if (pFlushGroup != NULL) {

×

2086

    pFlushList = tdListNew(POINTER_BYTES);

×

2088

  void*      pIte = NULL;

×

2089

  int32_t    iter = 0;

×

2090

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

×

2091

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

×

2092

    int32_t arraySize = TARRAY_SIZE(pWinStates);

×

2093

    if (minTs != INT64_MAX && arraySize > numOfKeep) {

×

2094

      SSessionKey key = {.win.skey = minTs, .win.ekey = minTs};

×

2095

      key.groupId = *(uint64_t*)tSimpleHashGetKey(pIte, NULL);

×

2096

      int32_t index = binarySearch(pWinStates, arraySize, &key, fillStateKeyCompare);

×

2097

      numOfKeep = TMAX(arraySize - index, MIN_NUM_OF_SORT_CACHE_WIN);

×

2098

      qDebug("modify numOfKeep, numOfKeep:%d. %s at line %d", numOfKeep, __func__, __LINE__);

×

2101

    int32_t size = arraySize - numOfKeep;

×

2102

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

×

2103

      SRowBuffPos* pPos = taosArrayGetP(pWinStates, i);

×

2104

      SSessionKey* pKey = pPos->pKey;

×

2105

      if (tSimpleHashGetSize(pFileState->pRecFlagMap) > 0) {

×

2106

        tSimpleHashRemove(pFileState->pRecFlagMap, pKey, sizeof(SSessionKey));

×

2108

      pPos->invalid = true;

×

2110

      if (i == 0 && pFlushGroup != NULL) {

×

2111

        void* pGpVal = tSimpleHashGet(pFlushGroup, &pKey->groupId, sizeof(uint64_t));

×

2112

        if (pGpVal == NULL) {

×

2113

          code = tdListAppend(pFlushList, &pPos);

×

2114

          QUERY_CHECK_CODE(code, lino, _end);

×

2115

          code = tSimpleHashPut(pFlushGroup, &pKey->groupId, sizeof(uint64_t), NULL, 0);

×

2116

          QUERY_CHECK_CODE(code, lino, _end);

×

2117

          continue;

×

2120

      pPos->beFlushed = true;

×

2121

      qTrace("clear expired session buff, ts:%" PRId64 ",groupid:%" PRIu64 ". %s at line %d", pKey->win.skey, pKey->groupId, __func__, __LINE__);

×

2123

      if (isFlushedState(pFileState, pKey->win.skey, 0)) {

×

2124

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

×

2125

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

×

2128

    taosArrayRemoveBatch(pWinStates, 0, size, NULL);

×

2131

  if (pFlushList != NULL) {

×

2132

    flushSnapshot(pFileState, pFlushList, false);

×

2133

    code = clearRowBuffNonFlush(pFileState);

×

2134

    QUERY_CHECK_CODE(code, lino, _end);

×

2135

    tdListFreeP(pFlushList, destroyRowBuffPosPtr);

×

2138

_end:

×

2139

  if (code != TSDB_CODE_SUCCESS) {

×

2140

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

×

2142

×

taosdata / TDengine / #3798

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

taosdata / TDengine / #3798

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