taosdata/TDengine | Build 4986 | source/dnode/vnode/src/meta/metaSnapshot.c | Coveralls

29

int32_t metaSnapReaderOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapReader** ppReader) {

×

30

  int32_t          code = 0;

×

32

  int32_t          c = 0;

×

33

  SMetaSnapReader* pReader = NULL;

×

36

  pReader = (SMetaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));

×

37

  if (pReader == NULL) {

×

38

    TSDB_CHECK_CODE(code = terrno, lino, _exit);

×

40

  pReader->pMeta = pMeta;

×

41

  pReader->sver = sver;

×

42

  pReader->ever = ever;

×

45

  code = tdbTbcOpen(pMeta->pTbDb, &pReader->pTbc, NULL);

×

46

  TSDB_CHECK_CODE(code, lino, _exit);

×

48

  code = tdbTbcMoveTo(pReader->pTbc, &(STbDbKey){.version = sver, .uid = INT64_MIN}, sizeof(STbDbKey), &c);

×

49

  TSDB_CHECK_CODE(code, lino, _exit);

×

51

_exit:

×

52

  if (code) {

×

53

    metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));

×

54

    metaSnapReaderClose(&pReader);

×

55

    *ppReader = NULL;

×

57

    metaInfo("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);

×

58

    *ppReader = pReader;

×

60

  return code;

×

63

void metaSnapReaderClose(SMetaSnapReader** ppReader) {

×

64

  if (ppReader && *ppReader) {

×

65

    tdbTbcClose((*ppReader)->pTbc);

×

66

    taosMemoryFree(*ppReader);

×

67

    *ppReader = NULL;

×

69

×

73

static int32_t metaDecodeEntryHeader(void* data, int32_t size, SMetaEntry* entry) {

×

74

  SDecoder decoder = {0};

×

75

  tDecoderInit(&decoder, (uint8_t*)data, size);

×

77

  int32_t code = metaDecodeEntryImpl(&decoder, entry, true);

×

78

  if (code) {

×

79

    tDecoderClear(&decoder);

×

80

    return code;

×

83

  tDecoderClear(&decoder);

×

84

  return 0;

×

87

int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {

×

88

  int32_t     code = 0;

×

89

  const void* pKey = NULL;

×

90

  const void* pData = NULL;

×

91

  int32_t     nKey = 0;

×

92

  int32_t     nData = 0;

×

96

  *ppData = NULL;

×

97

  while (pReader->iLoop < 2) {

×

98

    if (tdbTbcGet(pReader->pTbc, &pKey, &nKey, &pData, &nData) != 0 || ((STbDbKey*)pKey)->version > pReader->ever) {

×

99

      pReader->iLoop++;

×

102

      tdbTbcClose(pReader->pTbc);

×

103

      pReader->pTbc = NULL;

×

104

      code = tdbTbcOpen(pReader->pMeta->pTbDb, &pReader->pTbc, NULL);

×

105

      if (code) {

×

106

        metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pReader->pMeta->pVnode), __func__, __FILE__, __LINE__,

×

108

        goto _exit;

×

111

      code = tdbTbcMoveTo(pReader->pTbc, &(STbDbKey){.version = pReader->sver, .uid = INT64_MIN}, sizeof(STbDbKey), &c);

×

112

      if (code) {

×

113

        metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pReader->pMeta->pVnode), __func__, __FILE__, __LINE__,

×

115

        goto _exit;

×

118

      continue;

×

122

    SMetaEntry entry = {0};

×

123

    code = metaDecodeEntryHeader((void*)pData, nData, &entry);

×

124

    if (code) {

×

125

      metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pReader->pMeta->pVnode), __func__, __FILE__, __LINE__,

×

127

      goto _exit;

×

130

    key = ((STbDbKey*)pKey)[0];

×

131

    if (key.version < pReader->sver                                       //

×

132

        || (pReader->iLoop == 0 && TABS(entry.type) != TSDB_SUPER_TABLE)  // First loop send super table entry

×

133

        || (pReader->iLoop == 1 && TABS(entry.type) == TSDB_SUPER_TABLE)  // Second loop send non-super table entry

×

135

      if (tdbTbcMoveToNext(pReader->pTbc) != 0) {

×

136

        metaTrace("vgId:%d, vnode snapshot meta read data done", TD_VID(pReader->pMeta->pVnode));

×

138

      continue;

×

141

    if (!pData || !nData) {

×

142

      metaError("meta/snap: invalide nData: %" PRId32 " meta snap read failed.", nData);

×

143

      goto _exit;

×

146

    *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + nData);

×

147

    if (*ppData == NULL) {

×

148

      code = terrno;

×

149

      goto _exit;

×

152

    SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);

×

153

    pHdr->type = SNAP_DATA_META;

×

154

    pHdr->size = nData;

×

155

    memcpy(pHdr->data, pData, nData);

×

157

    metaDebug("vgId:%d, vnode snapshot meta read data, version:%" PRId64 " uid:%" PRId64 " blockLen:%d",

×

160

    if (tdbTbcMoveToNext(pReader->pTbc) != 0) {

×

161

      metaTrace("vgId:%d, vnode snapshot meta read data done", TD_VID(pReader->pMeta->pVnode));

×

163

    break;

×

166

_exit:

×

167

  if (code) {

×

168

    metaError("vgId:%d, vnode snapshot meta read data failed since %s", TD_VID(pReader->pMeta->pVnode),

×

171

  return code;

×

181

int32_t metaSnapWriterOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapWriter** ppWriter) {

×

182

  int32_t          code = 0;

×

187

  pWriter = (SMetaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));

×

188

  if (pWriter == NULL) {

×

189

    TSDB_CHECK_CODE(code = terrno, lino, _exit);

×

191

  pWriter->pMeta = pMeta;

×

192

  pWriter->sver = sver;

×

193

  pWriter->ever = ever;

×

195

  code = metaBegin(pMeta, META_BEGIN_HEAP_NIL);

×

196

  TSDB_CHECK_CODE(code, lino, _exit);

×

198

_exit:

×

199

  if (code) {

×

200

    metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));

×

201

    taosMemoryFree(pWriter);

×

202

    *ppWriter = NULL;

×

204

    metaDebug("vgId:%d, %s success", TD_VID(pMeta->pVnode), __func__);

×

205

    *ppWriter = pWriter;

×

207

  return code;

×

210

int32_t metaSnapWriterClose(SMetaSnapWriter** ppWriter, int8_t rollback) {

×

211

  int32_t          code = 0;

×

212

  SMetaSnapWriter* pWriter = *ppWriter;

×

214

  if (rollback) {

×

215

    metaInfo("vgId:%d, meta snapshot writer close and rollback start ", TD_VID(pWriter->pMeta->pVnode));

×

216

    code = metaAbort(pWriter->pMeta);

×

217

    metaInfo("vgId:%d, meta snapshot writer close and rollback finished, code:0x%x", TD_VID(pWriter->pMeta->pVnode),

×

219

    if (code) goto _err;

×

221

    code = metaCommit(pWriter->pMeta, pWriter->pMeta->txn);

×

222

    if (code) goto _err;

×

223

    code = metaFinishCommit(pWriter->pMeta, pWriter->pMeta->txn);

×

224

    if (code) goto _err;

×

226

  taosMemoryFree(pWriter);

×

227

  *ppWriter = NULL;

×

229

  return code;

×

231

_err:

×

232

  metaError("vgId:%d, meta snapshot writer close failed since %s", TD_VID(pWriter->pMeta->pVnode), tstrerror(code));

×

233

  return code;

×

236

int32_t metaSnapWrite(SMetaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {

×

237

  int32_t    code = 0;

×

238

  int32_t    lino = 0;

×

239

  SMeta*     pMeta = pWriter->pMeta;

×

240

  SMetaEntry metaEntry = {0};

×

241

  SDecoder*  pDecoder = &(SDecoder){0};

×

243

  tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));

×

244

  code = metaDecodeEntry(pDecoder, &metaEntry);

×

245

  TSDB_CHECK_CODE(code, lino, _exit);

×

247

  metaHandleSyncEntry(pMeta, &metaEntry);

×

249

_exit:

×

250

  if (code) {

×

251

    metaError("vgId:%d, %s failed at %s:%d since %s", TD_VID(pMeta->pVnode), __func__, __FILE__, lino, tstrerror(code));

×

253

  tDecoderClear(pDecoder);

×

254

  return code;

×

264

static void destroySTableInfoForChildTable(void* data) {

×

265

  STableInfoForChildTable* pData = (STableInfoForChildTable*)data;

×

266

  taosMemoryFree(pData->tableName);

×

267

  tDeleteSchemaWrapper(pData->schemaRow);

×

268

  tDeleteSchemaWrapper(pData->tagRow);

×

269

  taosMemoryFreeClear(pData->pExtSchemas);

×

270

×

272

static int32_t MoveToSnapShotVersion(SSnapContext* ctx) {

×

273

  int32_t code = 0;

×

274

  tdbTbcClose((TBC*)ctx->pCur);

×

275

  code = tdbTbcOpen(ctx->pMeta->pTbDb, (TBC**)&ctx->pCur, NULL);

×

276

  if (code != 0) {

×

277

    return TAOS_GET_TERRNO(code);

×

279

  STbDbKey key = {.version = ctx->snapVersion, .uid = INT64_MAX};

×

280

  int      c = 0;

×

281

  code = tdbTbcMoveTo((TBC*)ctx->pCur, &key, sizeof(key), &c);

×

282

  if (code != 0) {

×

283

    return TAOS_GET_TERRNO(code);

×

285

  if (c < 0) {

×

286

    if (tdbTbcMoveToPrev((TBC*)ctx->pCur) != 0) {

×

287

      metaTrace("vgId:%d, vnode snapshot move to prev failed", TD_VID(ctx->pMeta->pVnode));

×

290

  return 0;

×

293

static int32_t MoveToPosition(SSnapContext* ctx, int64_t ver, int64_t uid) {

×

294

  tdbTbcClose((TBC*)ctx->pCur);

×

295

  int32_t code = tdbTbcOpen(ctx->pMeta->pTbDb, (TBC**)&ctx->pCur, NULL);

×

296

  if (code != 0) {

×

297

    return TAOS_GET_TERRNO(code);

×

299

  STbDbKey key = {.version = ver, .uid = uid};

×

300

  int      c = 0;

×

301

  code = tdbTbcMoveTo((TBC*)ctx->pCur, &key, sizeof(key), &c);

×

302

  if (code != 0) {

×

303

    return TAOS_GET_TERRNO(code);

×

305

  return c;

×

308

static int32_t MoveToFirst(SSnapContext* ctx) {

×

309

  tdbTbcClose((TBC*)ctx->pCur);

×

310

  int32_t code = tdbTbcOpen(ctx->pMeta->pTbDb, (TBC**)&ctx->pCur, NULL);

×

311

  if (code != 0) {

×

312

    return TAOS_GET_TERRNO(code);

×

314

  code = tdbTbcMoveToFirst((TBC*)ctx->pCur);

×

315

  if (code != 0) {

×

316

    return TAOS_GET_TERRNO(code);

×

318

  return 0;

×

321

static int32_t saveSuperTableInfoForChildTable(SMetaEntry* me, SHashObj* suidInfo) {

×

322

  STableInfoForChildTable* data = (STableInfoForChildTable*)taosHashGet(suidInfo, &me->uid, sizeof(tb_uid_t));

×

323

  if (data) {

×

324

    return 0;

×

326

  int32_t                 code = 0;

×

327

  STableInfoForChildTable dataTmp = {0};

×

328

  dataTmp.tableName = taosStrdup(me->name);

×

329

  if (dataTmp.tableName == NULL) {

×

330

    code = terrno;

×

331

    goto END;

×

333

  dataTmp.schemaRow = tCloneSSchemaWrapper(&me->stbEntry.schemaRow);

×

334

  if (dataTmp.schemaRow == NULL) {

×

335

    code = TSDB_CODE_OUT_OF_MEMORY;

×

336

    goto END;

×

338

  dataTmp.tagRow = tCloneSSchemaWrapper(&me->stbEntry.schemaTag);

×

339

  if (dataTmp.tagRow == NULL) {

×

340

    code = TSDB_CODE_OUT_OF_MEMORY;

×

341

    goto END;

×

343

  if (me->pExtSchemas != NULL) {

×

344

    dataTmp.pExtSchemas = taosMemoryMalloc(sizeof(SExtSchema) * me->stbEntry.schemaRow.nCols);

×

345

    if (dataTmp.pExtSchemas == NULL) {

×

346

      code = TSDB_CODE_OUT_OF_MEMORY;

×

347

      goto END;

×

349

    memcpy(dataTmp.pExtSchemas, me->pExtSchemas, sizeof(SExtSchema) * me->stbEntry.schemaRow.nCols);

×

352

  code = taosHashPut(suidInfo, &me->uid, sizeof(tb_uid_t), &dataTmp, sizeof(STableInfoForChildTable));

×

353

  if (code != 0) {

×

354

    goto END;

×

356

  return 0;

×

358

END:

×

359

  destroySTableInfoForChildTable(&dataTmp);

×

360

  return TAOS_GET_TERRNO(code);

×

363

int32_t buildSnapContext(SVnode* pVnode, int64_t snapVersion, int64_t suid, int8_t subType, int8_t withMeta,

×

365

  int32_t code = 0;

×

366

  int32_t lino = 0;

×

367

  SDecoder   dc = {0};

×

368

  void* pKey = NULL;

×

369

  void* pVal = NULL;

×

370

  int   vLen = 0, kLen = 0;

×

372

  metaRLock(pVnode->pMeta);

×

373

  SSnapContext* ctx = taosMemoryCalloc(1, sizeof(SSnapContext));

×

374

  TSDB_CHECK_NULL(ctx, code, lino, END, terrno);

×

375

  *ctxRet = ctx;

×

376

  ctx->pMeta = pVnode->pMeta;

×

377

  ctx->snapVersion = snapVersion;

×

378

  ctx->suid = suid;

×

379

  ctx->subType = subType;

×

380

  ctx->queryMeta = withMeta;

×

381

  ctx->withMeta = withMeta;

×

382

  ctx->idVersion = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);

×

383

  TSDB_CHECK_NULL(ctx->idVersion, code, lino, END, terrno);

×

384

  ctx->suidInfo = taosHashInit(100, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);

×

385

  TSDB_CHECK_NULL(ctx->suidInfo, code, lino, END, terrno);

×

386

  taosHashSetFreeFp(ctx->suidInfo, destroySTableInfoForChildTable);

×

388

  ctx->index = 0;

×

389

  ctx->idList = taosArrayInit(100, sizeof(int64_t));

×

390

  TSDB_CHECK_NULL(ctx->idList, code, lino, END, terrno);

×

392

  metaDebug("tmqsnap init snapVersion:%" PRIi64, ctx->snapVersion);

×

393

  code = MoveToFirst(ctx);

×

394

  TSDB_CHECK_CODE(code, lino, END);

×

395

  while (1) {

×

396

    int32_t ret = tdbTbcNext((TBC*)ctx->pCur, &pKey, &kLen, &pVal, &vLen);

×

397

    if (ret < 0) break;

×

398

    STbDbKey* tmp = (STbDbKey*)pKey;

×

399

    if (tmp->version > ctx->snapVersion) break;

×

401

    SIdInfo* idData = (SIdInfo*)taosHashGet(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t));

×

402

    if (idData) {

×

403

      continue;

×

407

    if (tdbTbGet(ctx->pMeta->pUidIdx, &tmp->uid, sizeof(tb_uid_t), NULL, NULL) < 0) {

×

408

      continue;

×

411

    SMetaEntry me = {0};

×

412

    tDecoderInit(&dc, pVal, vLen);

×

413

    code = metaDecodeEntry(&dc, &me);

×

414

    TSDB_CHECK_CODE(code, lino, END);

×

415

    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {

×

416

      if (!((me.uid == ctx->suid && me.type == TSDB_SUPER_TABLE) ||

×

417

          (me.ctbEntry.suid == ctx->suid && (me.type == TSDB_CHILD_TABLE || me.type == TSDB_VIRTUAL_CHILD_TABLE)))) {

×

418

        tDecoderClear(&dc);

×

419

        continue;

×

423

    TSDB_CHECK_NULL(taosArrayPush(ctx->idList, &tmp->uid), code, lino, END, terrno);

×

424

    metaDebug("tmqsnap init idlist name:%s, uid:%" PRIi64, me.name, tmp->uid);

×

425

    tDecoderClear(&dc);

×

427

    SIdInfo info = {0};

×

428

    code = taosHashPut(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t), &info, sizeof(SIdInfo));

×

429

    TSDB_CHECK_CODE(code, lino, END);

×

431

  taosHashClear(ctx->idVersion);

×

433

  code = MoveToSnapShotVersion(ctx);

×

434

  TSDB_CHECK_CODE(code, lino, END);

×

436

  while (1) {

×

437

    int32_t ret = tdbTbcPrev((TBC*)ctx->pCur, &pKey, &kLen, &pVal, &vLen);

×

438

    if (ret < 0) break;

×

440

    STbDbKey* tmp = (STbDbKey*)pKey;

×

441

    SIdInfo*  idData = (SIdInfo*)taosHashGet(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t));

×

442

    if (idData) {

×

443

      continue;

×

445

    SIdInfo info = {.version = tmp->version, .index = 0};

×

446

    code = taosHashPut(ctx->idVersion, &tmp->uid, sizeof(tb_uid_t), &info, sizeof(SIdInfo));

×

447

    TSDB_CHECK_CODE(code, lino, END);

×

449

    SMetaEntry me = {0};

×

450

    tDecoderInit(&dc, pVal, vLen);

×

451

    code = metaDecodeEntry(&dc, &me);

×

452

    TSDB_CHECK_CODE(code, lino, END);

×

454

    if (ctx->subType == TOPIC_SUB_TYPE__TABLE) {

×

455

      if (!((me.uid == ctx->suid && me.type == TSDB_SUPER_TABLE) ||

×

456

          (me.ctbEntry.suid == ctx->suid && (me.type == TSDB_CHILD_TABLE || me.type == TSDB_VIRTUAL_CHILD_TABLE)))) {

×

457

        tDecoderClear(&dc);

×

458

        continue;

×

462

    if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_SUPER_TABLE) ||

×

463

        (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.uid == ctx->suid)) {

×

464

      code = saveSuperTableInfoForChildTable(&me, ctx->suidInfo);

×

465

      TSDB_CHECK_CODE(code, lino, END);

×

467

    tDecoderClear(&dc);

×

471

  for (int i = 0; i < taosArrayGetSize(ctx->idList); i++) {

×

472

    int64_t* uid = taosArrayGet(ctx->idList, i);

×

473

    TSDB_CHECK_NULL(uid, code, lino, END, terrno);

×

474

    SIdInfo* idData = (SIdInfo*)taosHashGet(ctx->idVersion, uid, sizeof(int64_t));

×

475

    TSDB_CHECK_NULL(idData, code, lino, END, terrno);

×

477

    idData->index = i;

×

478

    metaDebug("tmqsnap init idVersion uid:%" PRIi64 " version:%" PRIi64 " index:%d", *uid, idData->version, idData->index);

×

481

END:

×

482

  tdbFree(pKey);

×

483

  tdbFree(pVal);

×

484

  tDecoderClear(&dc);

×

486

  if (ctx != NULL) {

×

487

    tdbTbcClose((TBC*)ctx->pCur);

×

488

    ctx->pCur = NULL;

×

490

  metaULock(pVnode->pMeta);

×

492

  if(code != 0) {

×

493

    destroySnapContext(ctx);

×

494

    *ctxRet = NULL;

×

495

    metaError("tmqsnap build snap context failed line:%d since %s", lino, tstrerror(code));

×

497

  return code;

×

500

void destroySnapContext(SSnapContext* ctx) {

×

501

  if (ctx == NULL) {

×

502

    return;

×

504

  taosArrayDestroy(ctx->idList);

×

505

  taosHashCleanup(ctx->idVersion);

×

506

  taosHashCleanup(ctx->suidInfo);

×

507

  taosMemoryFree(ctx);

×

510

static int32_t buildNormalChildTableInfo(SVCreateTbReq* req, void** pBuf, int32_t* contLen) {

×

511

  int32_t            ret = 0;

×

512

  SVCreateTbBatchReq reqs = {0};

×

514

  reqs.pArray = taosArrayInit(1, sizeof(struct SVCreateTbReq));

×

515

  if (NULL == reqs.pArray) {

×

516

    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);

×

517

    goto end;

×

519

  if (taosArrayPush(reqs.pArray, req) == NULL) {

×

520

    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);

×

521

    goto end;

×

523

  reqs.nReqs = 1;

×

525

  tEncodeSize(tEncodeSVCreateTbBatchReq, &reqs, *contLen, ret);

×

526

  if (ret < 0) {

×

527

    ret = TAOS_GET_TERRNO(ret);

×

528

    goto end;

×

530

  *contLen += sizeof(SMsgHead);

×

531

  *pBuf = taosMemoryMalloc(*contLen);

×

532

  if (NULL == *pBuf) {

×

533

    ret = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);

×

534

    goto end;

×

536

  SEncoder coder = {0};

×

537

  tEncoderInit(&coder, POINTER_SHIFT(*pBuf, sizeof(SMsgHead)), *contLen);

×

538

  ret = tEncodeSVCreateTbBatchReq(&coder, &reqs);

×

539

  tEncoderClear(&coder);

×

541

  if (ret < 0) {

×

542

    taosMemoryFreeClear(*pBuf);

×

543

    ret = TAOS_GET_TERRNO(ret);

×

544

    goto end;

×

547

end:

×

548

  taosArrayDestroy(reqs.pArray);

×

549

  return ret;

×

552

static int32_t buildSuperTableInfo(SVCreateStbReq* req, void** pBuf, int32_t* contLen) {

×

553

  int32_t ret = 0;

×

554

  tEncodeSize(tEncodeSVCreateStbReq, req, *contLen, ret);

×

555

  if (ret < 0) {

×

556

    return TAOS_GET_TERRNO(ret);

×

559

  *contLen += sizeof(SMsgHead);

×

560

  *pBuf = taosMemoryMalloc(*contLen);

×

561

  if (NULL == *pBuf) {

×

562

    return TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);

×

565

  SEncoder encoder = {0};

×

566

  tEncoderInit(&encoder, POINTER_SHIFT(*pBuf, sizeof(SMsgHead)), *contLen);

×

567

  ret = tEncodeSVCreateStbReq(&encoder, req);

×

568

  tEncoderClear(&encoder);

×

569

  if (ret < 0) {

×

570

    taosMemoryFreeClear(*pBuf);

×

571

    return TAOS_GET_TERRNO(ret);

×

573

  return 0;

×

576

int32_t setForSnapShot(SSnapContext* ctx, int64_t uid) {

×

577

  if (uid == 0) {

×

578

    ctx->index = 0;

×

579

    return 0;

×

582

  SIdInfo* idInfo = (SIdInfo*)taosHashGet(ctx->idVersion, &uid, sizeof(tb_uid_t));

×

583

  if (idInfo == NULL) {

×

584

    return terrno;

×

587

  ctx->index = idInfo->index;

×

589

  return 0;

×

592

void taosXSetTablePrimaryKey(SSnapContext* ctx, int64_t uid) {

×

593

  bool            ret = false;

×

594

  SSchemaWrapper* schema = metaGetTableSchema(ctx->pMeta, uid, -1, 1, NULL, 0);

×

595

  if (schema && schema->nCols >= 2 && schema->pSchema[1].flags & COL_IS_KEY) {

×

596

    ret = true;

×

599

  ctx->hasPrimaryKey = ret;

×

600

×

602

bool taosXGetTablePrimaryKey(SSnapContext* ctx) { return ctx->hasPrimaryKey; }

×

604

int32_t getTableInfoFromSnapshot(SSnapContext* ctx, void** pBuf, int32_t* contLen, int16_t* type, int64_t* uid) {

×

605

  int32_t ret = 0;

×

606

  int32_t lino = 0;

×

607

  void*   pKey = NULL;

×

608

  void*   pVal = NULL;

×

609

  int     vLen = 0, kLen = 0;

×

610

  SDecoder   dc = {0};

×

611

  SArray* tagName = NULL;

×

612

  SArray* pTagVals = NULL;

×

614

  metaRLock(ctx->pMeta);

×

615

  while (1) {

×

616

    if (ctx->index >= taosArrayGetSize(ctx->idList)) {

×

617

      metaDebug("tmqsnap get meta end");

×

618

      ctx->index = 0;

×

619

      ctx->queryMeta = 0;  // change to get data

×

620

      goto END;

×

623

    int64_t* uidTmp = taosArrayGet(ctx->idList, ctx->index);

×

624

    TSDB_CHECK_NULL(uidTmp, ret, lino, END, terrno);

×

625

    ctx->index++;

×

626

    SIdInfo* idInfo = (SIdInfo*)taosHashGet(ctx->idVersion, uidTmp, sizeof(tb_uid_t));

×

627

    TSDB_CHECK_NULL(idInfo, ret, lino, END, terrno);

×

629

    *uid = *uidTmp;

×

630

    ret = MoveToPosition(ctx, idInfo->version, *uidTmp);

×

631

    if (ret == 0) {

×

632

      break;

×

634

    metaDebug("tmqsnap get meta not exist uid:%" PRIi64 " version:%" PRIi64, *uid, idInfo->version);

×

637

  ret = tdbTbcGet((TBC*)ctx->pCur, (const void**)&pKey, &kLen, (const void**)&pVal, &vLen);

×

638

  TSDB_CHECK_CONDITION(ret >= 0, ret, lino, END, TAOS_GET_TERRNO(ret));

×

639

  SMetaEntry me = {0};

×

640

  tDecoderInit(&dc, pVal, vLen);

×

641

  ret = metaDecodeEntry(&dc, &me);

×

642

  TSDB_CHECK_CONDITION(ret >= 0, ret, lino, END, TAOS_GET_TERRNO(ret));

×

643

  metaDebug("tmqsnap get meta uid:%" PRIi64 " name:%s index:%d", *uid, me.name, ctx->index - 1);

×

645

  if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_SUPER_TABLE) ||

×

646

      (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.uid == ctx->suid)) {

×

647

    SVCreateStbReq req = {0};

×

648

    req.name = me.name;

×

649

    req.suid = me.uid;

×

650

    req.schemaRow = me.stbEntry.schemaRow;

×

651

    req.schemaTag = me.stbEntry.schemaTag;

×

652

    req.schemaRow.version = 1;

×

653

    req.schemaTag.version = 1;

×

654

    req.colCmpr = me.colCmpr;

×

655

    req.pExtSchemas = me.pExtSchemas;

×

656

    req.virtualStb = TABLE_IS_VIRTUAL(me.flags);

×

658

    ret = buildSuperTableInfo(&req, pBuf, contLen);

×

659

    *type = TDMT_VND_CREATE_STB;

×

660

  } else if ((ctx->subType == TOPIC_SUB_TYPE__DB && (me.type == TSDB_CHILD_TABLE || me.type == TSDB_VIRTUAL_CHILD_TABLE)) ||

×

661

             (ctx->subType == TOPIC_SUB_TYPE__TABLE && (me.type == TSDB_CHILD_TABLE || me.type == TSDB_VIRTUAL_CHILD_TABLE) && me.ctbEntry.suid == ctx->suid)) {

×

663

        (STableInfoForChildTable*)taosHashGet(ctx->suidInfo, &me.ctbEntry.suid, sizeof(tb_uid_t));

×

664

    TSDB_CHECK_NULL(data, ret, lino, END, terrno);

×

666

    SVCreateTbReq req = {0};

×

668

    req.type = me.type;

×

669

    req.name = me.name;

×

670

    req.uid = me.uid;

×

671

    req.commentLen = -1;

×

672

    req.ctb.suid = me.ctbEntry.suid;

×

673

    req.ctb.tagNum = data->tagRow->nCols;

×

674

    req.ctb.stbName = data->tableName;

×

676

    tagName = taosArrayInit(req.ctb.tagNum, TSDB_COL_NAME_LEN);

×

677

    TSDB_CHECK_NULL(tagName, ret, lino, END, terrno);

×

678

    STag* p = (STag*)me.ctbEntry.pTags;

×

679

    if (tTagIsJson(p)) {

×

680

      if (p->nTag != 0) {

×

681

        SSchema* schema = &data->tagRow->pSchema[0];

×

682

        TSDB_CHECK_NULL(taosArrayPush(tagName, schema->name), ret, lino, END, terrno);

×

685

      ret = tTagToValArray((const STag*)p, &pTagVals);

×

686

      TSDB_CHECK_CODE(ret, lino, END);

×

687

      int16_t nCols = taosArrayGetSize(pTagVals);

×

688

      for (int j = 0; j < nCols; ++j) {

×

689

        STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);

×

690

        for (int i = 0; pTagVal && i < data->tagRow->nCols; i++) {

×

691

          SSchema* schema = &data->tagRow->pSchema[i];

×

692

          if (schema->colId == pTagVal->cid) {

×

693

            TSDB_CHECK_NULL(taosArrayPush(tagName, schema->name), ret, lino, END, terrno);

×

698

    req.ctb.pTag = me.ctbEntry.pTags;

×

699

    req.ctb.tagName = tagName;

×

700

    req.colRef = me.colRef;

×

701

    if (me.type == TSDB_VIRTUAL_CHILD_TABLE) {

×

702

      SMetaEntry *pSuper = NULL;

×

703

      int32_t code = metaFetchEntryByUid(ctx->pMeta, me.ctbEntry.suid, &pSuper);

×

704

      if (code == 0) {

×

705

        for (int i = 0; i < req.colRef.nCols && i < pSuper->stbEntry.schemaRow.nCols; i++) {

×

706

          SColRef *p = &req.colRef.pColRef[i];

×

707

          if (p->hasRef) {

×

708

            SSchema *schema = &pSuper->stbEntry.schemaRow.pSchema[i];

×

709

            tstrncpy(p->colName, schema->name, TSDB_COL_NAME_LEN);

×

713

      metaFetchEntryFree(&pSuper);

×

715

    ret = buildNormalChildTableInfo(&req, pBuf, contLen);

×

716

    *type = TDMT_VND_CREATE_TABLE;

×

717

  } else if (ctx->subType == TOPIC_SUB_TYPE__DB && (me.type == TSDB_NORMAL_TABLE || me.type == TSDB_VIRTUAL_NORMAL_TABLE)) {

×

718

    SVCreateTbReq req = {0};

×

719

    req.type = me.type;

×

720

    req.name = me.name;

×

721

    req.uid = me.uid;

×

722

    req.commentLen = -1;

×

723

    req.ntb.schemaRow = me.ntbEntry.schemaRow;

×

724

    req.colCmpr = me.colCmpr;

×

725

    req.pExtSchemas = me.pExtSchemas;

×

726

    req.colRef = me.colRef;

×

727

    ret = buildNormalChildTableInfo(&req, pBuf, contLen);

×

728

    *type = TDMT_VND_CREATE_TABLE;

×

730

    metaError("meta/snap: invalid topic sub type: %" PRId8 " get meta from snap failed.", ctx->subType);

×

731

    ret = TSDB_CODE_SDB_INVALID_TABLE_TYPE;

×

734

END:

×

735

  tdbTbcClose((TBC*)ctx->pCur);

×

736

  ctx->pCur = NULL;

×

737

  taosArrayDestroy(pTagVals);

×

738

  taosArrayDestroy(tagName);

×

739

  tDecoderClear(&dc);

×

740

  metaULock(ctx->pMeta);

×

742

  if(ret != 0) {

×

743

    metaError("tmqsnap get table info from snapshot failed line:%d since %s", lino, tstrerror(ret));

×

745

  return ret;

×

748

int32_t getMetaTableInfoFromSnapshot(SSnapContext* ctx, SMetaTableInfo* result) {

×

749

  void* pKey = NULL;

×

750

  void* pVal = NULL;

×

751

  int   vLen = 0;

×

752

  int   kLen = 0;

×

753

  int32_t code = 0;

×

754

  int32_t lino = 0;

×

755

  SDecoder   dc = {0};

×

757

  metaRLock(ctx->pMeta);

×

758

  while (1) {

×

759

    if (ctx->index >= taosArrayGetSize(ctx->idList)) {

×

760

      metaDebug("tmqsnap get uid info end");

×

761

      goto END;

×

763

    int64_t* uidTmp = taosArrayGet(ctx->idList, ctx->index);

×

764

    TSDB_CHECK_NULL(uidTmp, code, lino, END, terrno);

×

765

    ctx->index++;

×

766

    SIdInfo* idInfo = (SIdInfo*)taosHashGet(ctx->idVersion, uidTmp, sizeof(tb_uid_t));

×

767

    TSDB_CHECK_NULL(idInfo, code, lino, END, terrno);

×

769

    if (MoveToPosition(ctx, idInfo->version, *uidTmp) != 0) {

×

770

      metaDebug("tmqsnap getMetaTableInfoFromSnapshot not exist uid:%" PRIi64 " version:%" PRIi64, *uidTmp,

×

772

      continue;

×

774

    code = tdbTbcGet((TBC*)ctx->pCur, (const void**)&pKey, &kLen, (const void**)&pVal, &vLen);

×

775

    TSDB_CHECK_CODE(code, lino, END);

×

776

    SMetaEntry me = {0};

×

777

    tDecoderInit(&dc, pVal, vLen);

×

778

    code = metaDecodeEntry(&dc, &me);

×

779

    TSDB_CHECK_CODE(code, lino, END);

×

780

    metaDebug("tmqsnap get uid info uid:%" PRIi64 " name:%s index:%d", me.uid, me.name, ctx->index - 1);

×

782

    if ((ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_CHILD_TABLE) ||

×

783

        (ctx->subType == TOPIC_SUB_TYPE__TABLE && me.type == TSDB_CHILD_TABLE && me.ctbEntry.suid == ctx->suid)) {

×

785

          (STableInfoForChildTable*)taosHashGet(ctx->suidInfo, &me.ctbEntry.suid, sizeof(tb_uid_t));

×

786

      TSDB_CHECK_NULL(data, code, lino, END, terrno);

×

787

      result->suid = me.ctbEntry.suid;

×

788

      result->schema = tCloneSSchemaWrapper(data->schemaRow);

×

789

      if (data->pExtSchemas != NULL) {

×

790

        result->pExtSchemas = taosMemoryMalloc(sizeof(SExtSchema) * data->schemaRow->nCols);

×

791

        TSDB_CHECK_NULL(result->pExtSchemas, code, lino, END, terrno);

×

792

        memcpy(result->pExtSchemas, data->pExtSchemas, sizeof(SExtSchema) * data->schemaRow->nCols);

×

794

    } else if (ctx->subType == TOPIC_SUB_TYPE__DB && me.type == TSDB_NORMAL_TABLE) {

×

795

      result->suid = 0;

×

796

      result->schema = tCloneSSchemaWrapper(&me.ntbEntry.schemaRow);

×

797

      if (me.pExtSchemas != NULL) {

×

798

        result->pExtSchemas = taosMemoryMalloc(sizeof(SExtSchema) * me.ntbEntry.schemaRow.nCols);

×

799

        TSDB_CHECK_NULL(result->pExtSchemas, code, lino, END, terrno);

×

800

        memcpy(result->pExtSchemas, me.pExtSchemas, sizeof(SExtSchema) * me.ntbEntry.schemaRow.nCols);

×

803

      metaDebug("tmqsnap get uid continue");

×

804

      tDecoderClear(&dc);

×

805

      continue;

×

807

    result->uid = me.uid;

×

808

    tstrncpy(result->tbName, me.name, TSDB_TABLE_NAME_LEN);

×

809

    TSDB_CHECK_NULL(result->schema, code, lino, END, TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY));

×

810

    break;

×

813

END:

×

814

  tDecoderClear(&dc);

×

815

  tdbTbcClose((TBC*)ctx->pCur);

×

816

  ctx->pCur = NULL;

×

817

  metaULock(ctx->pMeta);

×

819

  if (code != 0) {

×

820

    metaError("tmqsnap get meta table info from snapshot failed line:%d since %s", lino, tstrerror(code));

×

822

  return code;

×

taosdata / TDengine / #4986

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

taosdata / TDengine / #4986

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