taosdata/TDengine | Build 3599 | source/dnode/vnode/src/meta/metaQuery.c | Coveralls

20

void _metaReaderInit(SMetaReader *pReader, void *pVnode, int32_t flags, SStoreMeta *pAPI) {

×

21

  SMeta *pMeta = ((SVnode *)pVnode)->pMeta;

×

22

  metaReaderDoInit(pReader, pMeta, flags);

×

23

  pReader->pAPI = pAPI;

×

24

×

26

void metaReaderDoInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags) {

×

27

  memset(pReader, 0, sizeof(*pReader));

×

28

  pReader->pMeta = pMeta;

×

29

  pReader->flags = flags;

×

30

  if (pReader->pMeta && !(flags & META_READER_NOLOCK)) {

×

31

    metaRLock(pMeta);

×

33

×

35

void metaReaderReleaseLock(SMetaReader *pReader) {

×

36

  if (pReader->pMeta && !(pReader->flags & META_READER_NOLOCK)) {

×

37

    metaULock(pReader->pMeta);

×

38

    pReader->flags |= META_READER_NOLOCK;

×

40

×

42

void metaReaderClear(SMetaReader *pReader) {

×

43

  if (pReader->pMeta && !(pReader->flags & META_READER_NOLOCK)) {

×

44

    metaULock(pReader->pMeta);

×

46

  tDecoderClear(&pReader->coder);

×

47

  tdbFree(pReader->pBuf);

×

48

  pReader->pBuf = NULL;

×

49

×

51

int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t uid) {

×

52

  int32_t  code = 0;

×

53

  SMeta   *pMeta = pReader->pMeta;

×

54

  STbDbKey tbDbKey = {.version = version, .uid = uid};

×

57

  if ((code = tdbTbGet(pMeta->pTbDb, &tbDbKey, sizeof(tbDbKey), &pReader->pBuf, &pReader->szBuf)) < 0) {

×

58

    return terrno = (TSDB_CODE_NOT_FOUND == code ? TSDB_CODE_PAR_TABLE_NOT_EXIST : code);

×

62

  tDecoderInit(&pReader->coder, pReader->pBuf, pReader->szBuf);

×

64

  code = metaDecodeEntry(&pReader->coder, &pReader->me);

×

65

  if (code) {

×

66

    tDecoderClear(&pReader->coder);

×

67

    return code;

×

71

  return 0;

×

74

bool metaIsTableExist(void *pVnode, tb_uid_t uid) {

×

75

  SVnode *pVnodeObj = pVnode;

×

76

  metaRLock(pVnodeObj->pMeta);  // query uid.idx

×

78

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

×

79

    metaULock(pVnodeObj->pMeta);

×

80

    return false;

×

83

  metaULock(pVnodeObj->pMeta);

×

84

  return true;

×

87

int metaReaderGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {

×

88

  SMeta  *pMeta = pReader->pMeta;

×

92

  if (tdbTbGet(pMeta->pUidIdx, &uid, sizeof(uid), &pReader->pBuf, &pReader->szBuf) < 0) {

×

93

    return terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

96

  version1 = ((SUidIdxVal *)pReader->pBuf)[0].version;

×

97

  return metaGetTableEntryByVersion(pReader, version1, uid);

×

100

int metaReaderGetTableEntryByUidCache(SMetaReader *pReader, tb_uid_t uid) {

×

101

  SMeta *pMeta = pReader->pMeta;

×

104

  int32_t   code = metaGetInfo(pMeta, uid, &info, pReader);

×

105

  if (TSDB_CODE_SUCCESS != code) {

×

106

    return terrno = (TSDB_CODE_NOT_FOUND == code ? TSDB_CODE_PAR_TABLE_NOT_EXIST : code);

×

109

  return metaGetTableEntryByVersion(pReader, info.version, uid);

×

112

int metaGetTableEntryByName(SMetaReader *pReader, const char *name) {

×

113

  SMeta   *pMeta = pReader->pMeta;

×

117

  if (tdbTbGet(pMeta->pNameIdx, name, strlen(name) + 1, &pReader->pBuf, &pReader->szBuf) < 0) {

×

118

    return terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

121

  uid = *(tb_uid_t *)pReader->pBuf;

×

122

  return metaReaderGetTableEntryByUid(pReader, uid);

×

125

tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) {

×

126

  void    *pData = NULL;

×

127

  int      nData = 0;

×

128

  tb_uid_t uid = 0;

×

130

  metaRLock(pMeta);

×

132

  if (tdbTbGet(pMeta->pNameIdx, name, strlen(name) + 1, &pData, &nData) == 0) {

×

133

    uid = *(tb_uid_t *)pData;

×

134

    tdbFree(pData);

×

137

  metaULock(pMeta);

×

139

  return uid;

×

142

int metaGetTableNameByUid(void *pVnode, uint64_t uid, char *tbName) {

×

143

  int         code = 0;

×

144

  SMetaReader mr = {0};

×

145

  metaReaderDoInit(&mr, ((SVnode *)pVnode)->pMeta, META_READER_LOCK);

×

146

  code = metaReaderGetTableEntryByUid(&mr, uid);

×

147

  if (code < 0) {

×

148

    metaReaderClear(&mr);

×

149

    return code;

×

152

  STR_TO_VARSTR(tbName, mr.me.name);

×

153

  metaReaderClear(&mr);

×

155

  return 0;

×

158

int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName) {

×

159

  int         code = 0;

×

160

  SMetaReader mr = {0};

×

161

  metaReaderDoInit(&mr, (SMeta *)meta, META_READER_LOCK);

×

162

  code = metaReaderGetTableEntryByUid(&mr, uid);

×

163

  if (code < 0) {

×

164

    metaReaderClear(&mr);

×

165

    return code;

×

167

  tstrncpy(tbName, mr.me.name, TSDB_TABLE_NAME_LEN);

×

168

  metaReaderClear(&mr);

×

170

  return 0;

×

173

int metaGetTableUidByName(void *pVnode, char *tbName, uint64_t *uid) {

×

174

  int         code = 0;

×

175

  SMetaReader mr = {0};

×

176

  metaReaderDoInit(&mr, ((SVnode *)pVnode)->pMeta, META_READER_LOCK);

×

178

  SMetaReader *pReader = &mr;

×

181

  if (tdbTbGet(((SMeta *)pReader->pMeta)->pNameIdx, tbName, strlen(tbName) + 1, &pReader->pBuf, &pReader->szBuf) < 0) {

×

182

    metaReaderClear(&mr);

×

183

    return terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

186

  *uid = *(tb_uid_t *)pReader->pBuf;

×

188

  metaReaderClear(&mr);

×

190

  return 0;

×

193

int metaGetTableTypeSuidByName(void *pVnode, char *tbName, ETableType *tbType, uint64_t* suid) {

×

194

  int         code = 0;

×

195

  SMetaReader mr = {0};

×

196

  metaReaderDoInit(&mr, ((SVnode *)pVnode)->pMeta, META_READER_LOCK);

×

198

  code = metaGetTableEntryByName(&mr, tbName);

×

199

  if (code == 0) *tbType = mr.me.type;

×

200

  if (TSDB_CHILD_TABLE == mr.me.type) {

×

201

    *suid = mr.me.ctbEntry.suid;

×

202

  } else if (TSDB_SUPER_TABLE == mr.me.type) {

×

203

    *suid = mr.me.uid;

×

205

    *suid = 0;

×

208

  metaReaderClear(&mr);

×

209

  return code;

×

212

int metaReadNext(SMetaReader *pReader) {

×

213

  SMeta *pMeta = pReader->pMeta;

×

217

  return 0;

×

220

int metaGetTableTtlByUid(void *meta, uint64_t uid, int64_t *ttlDays) {

×

221

  int         code = -1;

×

222

  SMetaReader mr = {0};

×

223

  metaReaderDoInit(&mr, (SMeta *)meta, META_READER_LOCK);

×

224

  code = metaReaderGetTableEntryByUid(&mr, uid);

×

225

  if (code < 0) {

×

226

    goto _exit;

×

228

  if (mr.me.type == TSDB_CHILD_TABLE) {

×

229

    *ttlDays = mr.me.ctbEntry.ttlDays;

×

230

  } else if (mr.me.type == TSDB_NORMAL_TABLE) {

×

231

    *ttlDays = mr.me.ntbEntry.ttlDays;

×

233

    goto _exit;

×

236

  code = 0;

×

238

_exit:

×

239

  metaReaderClear(&mr);

×

240

  return code;

×

244

SMTbCursor *metaOpenTbCursor(void *pVnode) {

×

245

  SMTbCursor *pTbCur = NULL;

×

248

  pTbCur = (SMTbCursor *)taosMemoryCalloc(1, sizeof(*pTbCur));

×

249

  if (pTbCur == NULL) {

×

250

    return NULL;

×

253

  SVnode *pVnodeObj = pVnode;

×

255

  pTbCur->pMeta = pVnodeObj->pMeta;

×

256

  pTbCur->paused = 1;

×

257

  code = metaResumeTbCursor(pTbCur, 1, 0);

×

258

  if (code) {

×

259

    terrno = code;

×

260

    taosMemoryFree(pTbCur);

×

261

    return NULL;

×

263

  return pTbCur;

×

266

void metaCloseTbCursor(SMTbCursor *pTbCur) {

×

267

  if (pTbCur) {

×

268

    tdbFree(pTbCur->pKey);

×

269

    tdbFree(pTbCur->pVal);

×

270

    if (!pTbCur->paused) {

×

271

      metaReaderClear(&pTbCur->mr);

×

272

      if (pTbCur->pDbc) {

×

273

        tdbTbcClose((TBC *)pTbCur->pDbc);

×

276

    taosMemoryFree(pTbCur);

×

278

×

280

void metaPauseTbCursor(SMTbCursor *pTbCur) {

×

281

  if (!pTbCur->paused) {

×

282

    metaReaderClear(&pTbCur->mr);

×

283

    tdbTbcClose((TBC *)pTbCur->pDbc);

×

284

    pTbCur->paused = 1;

×

286

×

287

int32_t metaResumeTbCursor(SMTbCursor *pTbCur, int8_t first, int8_t move) {

×

288

  int32_t code = 0;

×

290

  int8_t  locked = 0;

×

291

  if (pTbCur->paused) {

×

292

    metaReaderDoInit(&pTbCur->mr, pTbCur->pMeta, META_READER_LOCK);

×

293

    locked = 1;

×

294

    code = tdbTbcOpen(((SMeta *)pTbCur->pMeta)->pUidIdx, (TBC **)&pTbCur->pDbc, NULL);

×

295

    if (code != 0) {

×

296

      TSDB_CHECK_CODE(code, lino, _exit);

×

299

    if (first) {

×

300

      code = tdbTbcMoveToFirst((TBC *)pTbCur->pDbc);

×

301

      TSDB_CHECK_CODE(code, lino, _exit);

×

303

      int c = 1;

×

304

      code = tdbTbcMoveTo(pTbCur->pDbc, pTbCur->pKey, pTbCur->kLen, &c);

×

305

      TSDB_CHECK_CODE(code, lino, _exit);

×

306

      if (c == 0) {

×

307

        if (move) tdbTbcMoveToNext(pTbCur->pDbc);

×

308

      } else if (c < 0) {

×

309

        code = tdbTbcMoveToPrev(pTbCur->pDbc);

×

310

        TSDB_CHECK_CODE(code, lino, _exit);

×

312

        code = tdbTbcMoveToNext(pTbCur->pDbc);

×

313

        TSDB_CHECK_CODE(code, lino, _exit);

×

317

    pTbCur->paused = 0;

×

320

_exit:

×

321

  if (code != 0 && locked) {

×

322

    metaReaderReleaseLock(&pTbCur->mr);

×

324

  return code;

×

327

int32_t metaTbCursorNext(SMTbCursor *pTbCur, ETableType jumpTableType) {

×

333

    ret = tdbTbcNext((TBC *)pTbCur->pDbc, &pTbCur->pKey, &pTbCur->kLen, &pTbCur->pVal, &pTbCur->vLen);

×

334

    if (ret < 0) {

×

335

      return ret;

×

338

    tDecoderClear(&pTbCur->mr.coder);

×

340

    ret = metaGetTableEntryByVersion(&pTbCur->mr, ((SUidIdxVal *)pTbCur->pVal)[0].version, *(tb_uid_t *)pTbCur->pKey);

×

341

    if (ret) return ret;

×

343

    if (pTbCur->mr.me.type == jumpTableType) {

×

344

      continue;

×

347

    break;

×

350

  return 0;

×

353

int32_t metaTbCursorPrev(SMTbCursor *pTbCur, ETableType jumpTableType) {

×

359

    ret = tdbTbcPrev((TBC *)pTbCur->pDbc, &pTbCur->pKey, &pTbCur->kLen, &pTbCur->pVal, &pTbCur->vLen);

×

360

    if (ret < 0) {

×

361

      return -1;

×

364

    tDecoderClear(&pTbCur->mr.coder);

×

366

    ret = metaGetTableEntryByVersion(&pTbCur->mr, ((SUidIdxVal *)pTbCur->pVal)[0].version, *(tb_uid_t *)pTbCur->pKey);

×

367

    if (ret < 0) {

×

368

      return ret;

×

371

    if (pTbCur->mr.me.type == jumpTableType) {

×

372

      continue;

×

375

    break;

×

378

  return 0;

×

381

SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock, int64_t *createTime) {

×

382

  void           *pData = NULL;

×

383

  int             nData = 0;

×

385

  SSchemaWrapper  schema = {0};

×

386

  SSchemaWrapper *pSchema = NULL;

×

387

  SDecoder        dc = {0};

×

388

  if (lock) {

×

389

    metaRLock(pMeta);

×

391

_query:

×

392

  if (tdbTbGet(pMeta->pUidIdx, &uid, sizeof(uid), &pData, &nData) < 0) {

×

393

    goto _err;

×

396

  version = ((SUidIdxVal *)pData)[0].version;

×

398

  if (tdbTbGet(pMeta->pTbDb, &(STbDbKey){.uid = uid, .version = version}, sizeof(STbDbKey), &pData, &nData) != 0) {

×

399

    goto _err;

×

402

  SMetaEntry me = {0};

×

403

  tDecoderInit(&dc, pData, nData);

×

404

  int32_t code = metaDecodeEntry(&dc, &me);

×

405

  if (code) {

×

406

    tDecoderClear(&dc);

×

407

    goto _err;

×

409

  if (me.type == TSDB_SUPER_TABLE) {

×

410

    if (sver == -1 || sver == me.stbEntry.schemaRow.version) {

×

411

      pSchema = tCloneSSchemaWrapper(&me.stbEntry.schemaRow);

×

412

      tDecoderClear(&dc);

×

413

      goto _exit;

×

415

  } else if (me.type == TSDB_CHILD_TABLE) {

×

416

    uid = me.ctbEntry.suid;

×

417

    if (createTime != NULL){

×

418

      *createTime = me.ctbEntry.btime;

×

420

    tDecoderClear(&dc);

×

421

    goto _query;

×

423

    if (sver == -1 || sver == me.ntbEntry.schemaRow.version) {

×

424

      pSchema = tCloneSSchemaWrapper(&me.ntbEntry.schemaRow);

×

425

      tDecoderClear(&dc);

×

426

      goto _exit;

×

429

  tDecoderClear(&dc);

×

432

  if (tdbTbGet(pMeta->pSkmDb, &(SSkmDbKey){.uid = uid, .sver = sver}, sizeof(SSkmDbKey), &pData, &nData) < 0) {

×

433

    goto _err;

×

436

  tDecoderInit(&dc, pData, nData);

×

437

  if (tDecodeSSchemaWrapperEx(&dc, &schema) != 0) {

×

438

    goto _err;

×

440

  pSchema = tCloneSSchemaWrapper(&schema);

×

441

  tDecoderClear(&dc);

×

443

_exit:

×

444

  if (lock) {

×

445

    metaULock(pMeta);

×

447

  tdbFree(pData);

×

448

  return pSchema;

×

450

_err:

×

451

  if (lock) {

×

452

    metaULock(pMeta);

×

454

  tdbFree(pData);

×

455

  return NULL;

×

458

SMCtbCursor *metaOpenCtbCursor(void *pVnode, tb_uid_t uid, int lock) {

×

459

  SMeta       *pMeta = ((SVnode *)pVnode)->pMeta;

×

460

  SMCtbCursor *pCtbCur = NULL;

×

462

  int          ret = 0;

×

463

  int          c = 0;

×

465

  pCtbCur = (SMCtbCursor *)taosMemoryCalloc(1, sizeof(*pCtbCur));

×

466

  if (pCtbCur == NULL) {

×

467

    return NULL;

×

470

  pCtbCur->pMeta = pMeta;

×

471

  pCtbCur->suid = uid;

×

472

  pCtbCur->lock = lock;

×

473

  pCtbCur->paused = 1;

×

475

  ret = metaResumeCtbCursor(pCtbCur, 1);

×

476

  if (ret < 0) {

×

477

    return NULL;

×

479

  return pCtbCur;

×

482

void metaCloseCtbCursor(SMCtbCursor *pCtbCur) {

×

483

  if (pCtbCur) {

×

484

    if (!pCtbCur->paused) {

×

485

      if (pCtbCur->pMeta && pCtbCur->lock) metaULock(pCtbCur->pMeta);

×

486

      if (pCtbCur->pCur) {

×

487

        tdbTbcClose(pCtbCur->pCur);

×

490

    tdbFree(pCtbCur->pKey);

×

491

    tdbFree(pCtbCur->pVal);

×

493

  taosMemoryFree(pCtbCur);

×

494

×

496

void metaPauseCtbCursor(SMCtbCursor *pCtbCur) {

×

497

  if (!pCtbCur->paused) {

×

498

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

×

499

    if (pCtbCur->lock) {

×

500

      metaULock(pCtbCur->pMeta);

×

502

    pCtbCur->paused = 1;

×

504

×

506

int32_t metaResumeCtbCursor(SMCtbCursor *pCtbCur, int8_t first) {

×

507

  if (pCtbCur->paused) {

×

508

    pCtbCur->paused = 0;

×

510

    if (pCtbCur->lock) {

×

511

      metaRLock(pCtbCur->pMeta);

×

513

    int ret = 0;

×

514

    ret = tdbTbcOpen(pCtbCur->pMeta->pCtbIdx, (TBC **)&pCtbCur->pCur, NULL);

×

515

    if (ret < 0) {

×

516

      metaCloseCtbCursor(pCtbCur);

×

517

      return -1;

×

520

    if (first) {

×

523

      ctbIdxKey.suid = pCtbCur->suid;

×

524

      ctbIdxKey.uid = INT64_MIN;

×

525

      int c = 0;

×

526

      ret = tdbTbcMoveTo(pCtbCur->pCur, &ctbIdxKey, sizeof(ctbIdxKey), &c);

×

527

      if (c > 0) {

×

528

        ret = tdbTbcMoveToNext(pCtbCur->pCur);

×

531

      int c = 0;

×

532

      ret = tdbTbcMoveTo(pCtbCur->pCur, pCtbCur->pKey, pCtbCur->kLen, &c);

×

533

      if (c < 0) {

×

534

        ret = tdbTbcMoveToPrev(pCtbCur->pCur);

×

536

        ret = tdbTbcMoveToNext(pCtbCur->pCur);

×

540

  return 0;

×

543

tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {

×

547

  ret = tdbTbcNext(pCtbCur->pCur, &pCtbCur->pKey, &pCtbCur->kLen, &pCtbCur->pVal, &pCtbCur->vLen);

×

548

  if (ret < 0) {

×

549

    return 0;

×

552

  pCtbIdxKey = pCtbCur->pKey;

×

553

  if (pCtbIdxKey->suid > pCtbCur->suid) {

×

554

    return 0;

×

557

  return pCtbIdxKey->uid;

×

570

SMStbCursor *metaOpenStbCursor(SMeta *pMeta, tb_uid_t suid) {

×

571

  SMStbCursor *pStbCur = NULL;

×

572

  int          ret = 0;

×

573

  int          c = 0;

×

575

  pStbCur = (SMStbCursor *)taosMemoryCalloc(1, sizeof(*pStbCur));

×

576

  if (pStbCur == NULL) {

×

577

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

578

    return NULL;

×

581

  pStbCur->pMeta = pMeta;

×

582

  pStbCur->suid = suid;

×

583

  metaRLock(pMeta);

×

585

  ret = tdbTbcOpen(pMeta->pSuidIdx, &pStbCur->pCur, NULL);

×

586

  if (ret < 0) {

×

587

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

588

    metaULock(pMeta);

×

589

    taosMemoryFree(pStbCur);

×

590

    return NULL;

×

594

  ret = tdbTbcMoveTo(pStbCur->pCur, &suid, sizeof(suid), &c);

×

595

  if (c > 0) {

×

596

    ret = tdbTbcMoveToNext(pStbCur->pCur);

×

599

  return pStbCur;

×

602

void metaCloseStbCursor(SMStbCursor *pStbCur) {

×

603

  if (pStbCur) {

×

604

    if (pStbCur->pMeta) metaULock(pStbCur->pMeta);

×

605

    if (pStbCur->pCur) {

×

606

      tdbTbcClose(pStbCur->pCur);

×

608

      tdbFree(pStbCur->pKey);

×

609

      tdbFree(pStbCur->pVal);

×

612

    taosMemoryFree(pStbCur);

×

614

×

616

tb_uid_t metaStbCursorNext(SMStbCursor *pStbCur) {

×

619

  ret = tdbTbcNext(pStbCur->pCur, &pStbCur->pKey, &pStbCur->kLen, &pStbCur->pVal, &pStbCur->vLen);

×

620

  if (ret < 0) {

×

621

    return 0;

×

623

  return *(tb_uid_t *)pStbCur->pKey;

×

626

STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock) {

×

627

  STSchema       *pTSchema = NULL;

×

628

  SSchemaWrapper *pSW = NULL;

×

630

  pSW = metaGetTableSchema(pMeta, uid, sver, lock, NULL);

×

631

  if (!pSW) return NULL;

×

633

  pTSchema = tBuildTSchema(pSW->pSchema, pSW->nCols, pSW->version);

×

635

  taosMemoryFree(pSW->pSchema);

×

636

  taosMemoryFree(pSW);

×

637

  return pTSchema;

×

640

int32_t metaGetTbTSchemaNotNull(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock, STSchema **ppTSchema) {

×

641

  *ppTSchema = metaGetTbTSchema(pMeta, uid, sver, lock);

×

642

  if (*ppTSchema == NULL) {

×

643

    return terrno;

×

645

  return TSDB_CODE_SUCCESS;

×

648

int32_t metaGetTbTSchemaMaybeNull(SMeta *pMeta, tb_uid_t uid, int32_t sver, int lock, STSchema **ppTSchema) {

×

649

  *ppTSchema = metaGetTbTSchema(pMeta, uid, sver, lock);

×

650

  if (*ppTSchema == NULL && terrno == TSDB_CODE_OUT_OF_MEMORY) {

×

651

    return terrno;

×

653

  return TSDB_CODE_SUCCESS;

×

656

int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema **ppTSchema) {

×

657

  int32_t code = 0;

×

660

  void     *pData = NULL;

×

661

  int       nData = 0;

×

663

  if (sver <= 0) {

×

665

    if (metaGetInfo(pMeta, suid ? suid : uid, &info, NULL) == 0) {

×

666

      sver = info.skmVer;

×

668

      TBC *pSkmDbC = NULL;

×

671

      skmDbKey.uid = suid ? suid : uid;

×

672

      skmDbKey.sver = INT32_MAX;

×

674

      code = tdbTbcOpen(pMeta->pSkmDb, &pSkmDbC, NULL);

×

675

      TSDB_CHECK_CODE(code, lino, _exit);

×

676

      metaRLock(pMeta);

×

678

      if (tdbTbcMoveTo(pSkmDbC, &skmDbKey, sizeof(skmDbKey), &c) < 0) {

×

679

        metaULock(pMeta);

×

680

        tdbTbcClose(pSkmDbC);

×

681

        code = TSDB_CODE_NOT_FOUND;

×

682

        goto _exit;

×

685

      if (c == 0) {

×

686

        metaULock(pMeta);

×

687

        tdbTbcClose(pSkmDbC);

×

688

        code = TSDB_CODE_FAILED;

×

689

        metaError("meta/query: incorrect c: %" PRId32 ".", c);

×

690

        goto _exit;

×

693

      if (c < 0) {

×

694

        int32_t ret = tdbTbcMoveToPrev(pSkmDbC);

×

697

      const void *pKey = NULL;

×

698

      int32_t     nKey = 0;

×

699

      int32_t     ret = tdbTbcGet(pSkmDbC, &pKey, &nKey, NULL, NULL);

×

701

      if (((SSkmDbKey *)pKey)->uid != skmDbKey.uid) {

×

702

        metaULock(pMeta);

×

703

        tdbTbcClose(pSkmDbC);

×

704

        code = TSDB_CODE_NOT_FOUND;

×

705

        goto _exit;

×

708

      sver = ((SSkmDbKey *)pKey)->sver;

×

710

      metaULock(pMeta);

×

711

      tdbTbcClose(pSkmDbC);

×

715

  if (!(sver > 0)) {

×

716

    code = TSDB_CODE_NOT_FOUND;

×

717

    goto _exit;

×

720

  skmDbKey.uid = suid ? suid : uid;

×

721

  skmDbKey.sver = sver;

×

722

  metaRLock(pMeta);

×

723

  if (tdbTbGet(pMeta->pSkmDb, &skmDbKey, sizeof(SSkmDbKey), &pData, &nData) < 0) {

×

724

    metaULock(pMeta);

×

725

    code = TSDB_CODE_NOT_FOUND;

×

726

    goto _exit;

×

728

  metaULock(pMeta);

×

731

  SDecoder        dc = {0};

×

733

  SSchemaWrapper *pSchemaWrapper = &schema;

×

735

  tDecoderInit(&dc, pData, nData);

×

736

  code = tDecodeSSchemaWrapper(&dc, pSchemaWrapper);

×

737

  tDecoderClear(&dc);

×

738

  tdbFree(pData);

×

739

  if (TSDB_CODE_SUCCESS != code) {

×

740

    taosMemoryFree(pSchemaWrapper->pSchema);

×

741

    goto _exit;

×

745

  STSchema *pTSchema = tBuildTSchema(pSchemaWrapper->pSchema, pSchemaWrapper->nCols, pSchemaWrapper->version);

×

746

  if (pTSchema == NULL) {

×

747

    code = TSDB_CODE_OUT_OF_MEMORY;

×

750

  *ppTSchema = pTSchema;

×

751

  taosMemoryFree(pSchemaWrapper->pSchema);

×

753

_exit:

×

754

  return code;

×

758

int64_t metaGetTbNum(SMeta *pMeta) {

×

764

  return pMeta->pVnode->config.vndStats.numOfCTables + pMeta->pVnode->config.vndStats.numOfNTables;

×

767

void metaUpdTimeSeriesNum(SMeta *pMeta) {

×

768

  int64_t nCtbTimeSeries = 0;

×

769

  if (vnodeGetTimeSeriesNum(pMeta->pVnode, &nCtbTimeSeries) == 0) {

×

770

    atomic_store_64(&pMeta->pVnode->config.vndStats.numOfTimeSeries, nCtbTimeSeries);

×

772

×

776

  SVnodeStats *pStats = &pMeta->pVnode->config.vndStats;

×

777

  if (forceUpd || pStats->numOfTimeSeries <= 0) {

×

778

    metaUpdTimeSeriesNum(pMeta);

×

781

  return pStats->numOfTimeSeries + pStats->numOfNTimeSeries;

×

785

int64_t metaGetTimeSeriesNum(SMeta *pMeta, int type) {

×

786

  int64_t nTimeSeries = metaGetTimeSeriesNumImpl(pMeta, false);

×

787

  if (type == 1) {

×

788

    atomic_store_64(&pMeta->pVnode->config.vndStats.numOfReportedTimeSeries, nTimeSeries);

×

790

  return nTimeSeries;

×

803

SMSmaCursor *metaOpenSmaCursor(SMeta *pMeta, tb_uid_t uid) {

×

804

  SMSmaCursor *pSmaCur = NULL;

×

809

  pSmaCur = (SMSmaCursor *)taosMemoryCalloc(1, sizeof(*pSmaCur));

×

810

  if (pSmaCur == NULL) {

×

811

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

812

    return NULL;

×

815

  pSmaCur->pMeta = pMeta;

×

816

  pSmaCur->uid = uid;

×

817

  metaRLock(pMeta);

×

819

  ret = tdbTbcOpen(pMeta->pSmaIdx, &pSmaCur->pCur, NULL);

×

820

  if (ret < 0) {

×

821

    metaULock(pMeta);

×

822

    taosMemoryFree(pSmaCur);

×

823

    return NULL;

×

827

  smaIdxKey.uid = uid;

×

828

  smaIdxKey.smaUid = INT64_MIN;

×

829

  ret = tdbTbcMoveTo(pSmaCur->pCur, &smaIdxKey, sizeof(smaIdxKey), &c);

×

830

  if (c > 0) {

×

831

    ret = tdbTbcMoveToNext(pSmaCur->pCur);

×

834

  return pSmaCur;

×

837

void metaCloseSmaCursor(SMSmaCursor *pSmaCur) {

×

838

  if (pSmaCur) {

×

839

    if (pSmaCur->pMeta) metaULock(pSmaCur->pMeta);

×

840

    if (pSmaCur->pCur) {

×

841

      tdbTbcClose(pSmaCur->pCur);

×

842

      pSmaCur->pCur = NULL;

×

844

      tdbFree(pSmaCur->pKey);

×

845

      tdbFree(pSmaCur->pVal);

×

848

    taosMemoryFree(pSmaCur);

×

850

×

852

tb_uid_t metaSmaCursorNext(SMSmaCursor *pSmaCur) {

×

856

  ret = tdbTbcNext(pSmaCur->pCur, &pSmaCur->pKey, &pSmaCur->kLen, &pSmaCur->pVal, &pSmaCur->vLen);

×

857

  if (ret < 0) {

×

858

    return 0;

×

861

  pSmaIdxKey = pSmaCur->pKey;

×

862

  if (pSmaIdxKey->uid > pSmaCur->uid) {

×

863

    return 0;

×

866

  return pSmaIdxKey->uid;

×

869

STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) {

×

870

  STSmaWrapper *pSW = NULL;

×

871

  SArray       *pSmaIds = NULL;

×

873

  if (!(pSmaIds = metaGetSmaIdsByTable(pMeta, uid))) {

×

874

    return NULL;

×

877

  pSW = taosMemoryCalloc(1, sizeof(*pSW));

×

878

  if (!pSW) {

×

879

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

880

    goto _err;

×

883

  pSW->number = taosArrayGetSize(pSmaIds);

×

884

  pSW->tSma = taosMemoryCalloc(pSW->number, sizeof(STSma));

×

886

  if (!pSW->tSma) {

×

887

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

888

    goto _err;

×

891

  SMetaReader mr = {0};

×

892

  metaReaderDoInit(&mr, pMeta, META_READER_LOCK);

×

894

  int     smaIdx = 0;

×

895

  STSma  *pTSma = NULL;

×

896

  for (int i = 0; i < pSW->number; ++i) {

×

897

    smaId = *(tb_uid_t *)taosArrayGet(pSmaIds, i);

×

898

    if (metaReaderGetTableEntryByUid(&mr, smaId) < 0) {

×

899

      tDecoderClear(&mr.coder);

×

900

      metaWarn("vgId:%d, no entry for tbId:%" PRIi64 ", smaId:%" PRIi64, TD_VID(pMeta->pVnode), uid, smaId);

×

901

      continue;

×

903

    tDecoderClear(&mr.coder);

×

904

    pTSma = pSW->tSma + smaIdx;

×

905

    memcpy(pTSma, mr.me.smaEntry.tsma, sizeof(STSma));

×

906

    if (deepCopy) {

×

907

      if (pTSma->exprLen > 0) {

×

908

        if (!(pTSma->expr = taosMemoryCalloc(1, pTSma->exprLen))) {

×

909

          terrno = TSDB_CODE_OUT_OF_MEMORY;

×

910

          goto _err;

×

912

        memcpy((void *)pTSma->expr, mr.me.smaEntry.tsma->expr, pTSma->exprLen);

×

914

      if (pTSma->tagsFilterLen > 0) {

×

915

        if (!(pTSma->tagsFilter = taosMemoryCalloc(1, pTSma->tagsFilterLen))) {

×

916

          terrno = TSDB_CODE_OUT_OF_MEMORY;

×

917

          goto _err;

×

920

      memcpy((void *)pTSma->tagsFilter, mr.me.smaEntry.tsma->tagsFilter, pTSma->tagsFilterLen);

×

922

      pTSma->exprLen = 0;

×

923

      pTSma->expr = NULL;

×

924

      pTSma->tagsFilterLen = 0;

×

925

      pTSma->tagsFilter = NULL;

×

928

    ++smaIdx;

×

931

  if (smaIdx <= 0) goto _err;

×

932

  pSW->number = smaIdx;

×

934

  metaReaderClear(&mr);

×

935

  taosArrayDestroy(pSmaIds);

×

936

  return pSW;

×

937

_err:

×

938

  metaReaderClear(&mr);

×

939

  taosArrayDestroy(pSmaIds);

×

940

  pSW = tFreeTSmaWrapper(pSW, deepCopy);

×

941

  return NULL;

×

944

STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {

×

945

  STSma      *pTSma = NULL;

×

946

  SMetaReader mr = {0};

×

947

  metaReaderDoInit(&mr, pMeta, META_READER_LOCK);

×

948

  if (metaReaderGetTableEntryByUid(&mr, indexUid) < 0) {

×

949

    metaWarn("vgId:%d, failed to get table entry for smaId:%" PRIi64, TD_VID(pMeta->pVnode), indexUid);

×

950

    metaReaderClear(&mr);

×

951

    return NULL;

×

953

  pTSma = (STSma *)taosMemoryMalloc(sizeof(STSma));

×

954

  if (!pTSma) {

×

955

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

956

    metaReaderClear(&mr);

×

957

    return NULL;

×

960

  memcpy(pTSma, mr.me.smaEntry.tsma, sizeof(STSma));

×

962

  metaReaderClear(&mr);

×

963

  return pTSma;

×

966

SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) {

×

967

  SArray     *pUids = NULL;

×

968

  SSmaIdxKey *pSmaIdxKey = NULL;

×

970

  SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid);

×

971

  if (!pCur) {

×

972

    return NULL;

×

975

  while (1) {

×

976

    tb_uid_t id = metaSmaCursorNext(pCur);

×

977

    if (id == 0) {

×

978

      break;

×

981

    if (!pUids) {

×

982

      pUids = taosArrayInit(16, sizeof(tb_uid_t));

×

983

      if (!pUids) {

×

984

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

985

        metaCloseSmaCursor(pCur);

×

986

        return NULL;

×

990

    pSmaIdxKey = (SSmaIdxKey *)pCur->pKey;

×

992

    if (!taosArrayPush(pUids, &pSmaIdxKey->smaUid)) {

×

993

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

994

      metaCloseSmaCursor(pCur);

×

995

      taosArrayDestroy(pUids);

×

996

      return NULL;

×

1000

  metaCloseSmaCursor(pCur);

×

1001

  return pUids;

×

1004

SArray *metaGetSmaTbUids(SMeta *pMeta) {

×

1005

  SArray     *pUids = NULL;

×

1006

  SSmaIdxKey *pSmaIdxKey = NULL;

×

1007

  tb_uid_t    lastUid = 0;

×

1009

  SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, 0);

×

1010

  if (!pCur) {

×

1011

    return NULL;

×

1014

  while (1) {

×

1015

    tb_uid_t uid = metaSmaCursorNext(pCur);

×

1016

    if (uid == 0) {

×

1017

      break;

×

1020

    if (lastUid == uid) {

×

1021

      continue;

×

1024

    lastUid = uid;

×

1026

    if (!pUids) {

×

1027

      pUids = taosArrayInit(16, sizeof(tb_uid_t));

×

1028

      if (!pUids) {

×

1029

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1030

        metaCloseSmaCursor(pCur);

×

1031

        return NULL;

×

1035

    if (!taosArrayPush(pUids, &uid)) {

×

1036

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1037

      metaCloseSmaCursor(pCur);

×

1038

      taosArrayDestroy(pUids);

×

1039

      return NULL;

×

1043

  metaCloseSmaCursor(pCur);

×

1044

  return pUids;

×

1049

const void *metaGetTableTagVal(const void *pTag, int16_t type, STagVal *val) {

×

1050

  STag *tag = (STag *)pTag;

×

1051

  if (type == TSDB_DATA_TYPE_JSON) {

×

1052

    return tag;

×

1054

  bool find = tTagGet(tag, val);

×

1056

  if (!find) {

×

1057

    return NULL;

×

1060

  return val;

×

1075

int32_t metaFilterCreateTime(void *pVnode, SMetaFltParam *arg, SArray *pUids) {

×

1076

  SMeta         *pMeta = ((SVnode *)pVnode)->pMeta;

×

1077

  SMetaFltParam *param = arg;

×

1078

  int32_t        ret = 0;

×

1080

  SIdxCursor *pCursor = NULL;

×

1081

  pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));

×

1082

  if (pCursor == NULL) {

×

1083

    return terrno;

×

1085

  pCursor->pMeta = pMeta;

×

1086

  pCursor->suid = param->suid;

×

1087

  pCursor->cid = param->cid;

×

1088

  pCursor->type = param->type;

×

1090

  metaRLock(pMeta);

×

1091

  ret = tdbTbcOpen(pMeta->pBtimeIdx, &pCursor->pCur, NULL);

×

1092

  if (ret != 0) {

×

1093

    goto END;

×

1095

  int64_t uidLimit = param->reverse ? INT64_MAX : 0;

×

1097

  SBtimeIdxKey  btimeKey = {.btime = *(int64_t *)(param->val), .uid = uidLimit};

×

1098

  SBtimeIdxKey *pBtimeKey = &btimeKey;

×

1100

  int cmp = 0;

×

1101

  if (tdbTbcMoveTo(pCursor->pCur, &btimeKey, sizeof(btimeKey), &cmp) < 0) {

×

1102

    goto END;

×

1105

  int32_t valid = 0;

×

1106

  int32_t count = 0;

×

1109

  do {

×

1110

    void   *entryKey = NULL;

×

1111

    int32_t nEntryKey = -1;

×

1112

    valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, NULL, NULL);

×

1113

    if (valid < 0) break;

×

1115

    SBtimeIdxKey *p = entryKey;

×

1116

    if (count > TRY_ERROR_LIMIT) break;

×

1118

    terrno = TSDB_CODE_SUCCESS;

×

1119

    int32_t cmp = (*param->filterFunc)((void *)&p->btime, (void *)&pBtimeKey->btime, param->type);

×

1120

    if (terrno != TSDB_CODE_SUCCESS) {

×

1121

      ret = terrno;

×

1122

      break;

×

1124

    if (cmp == 0) {

×

1125

      if (taosArrayPush(pUids, &p->uid) == NULL) {

×

1126

        ret = terrno;

×

1127

        break;

×

1130

      if (param->equal == true) {

×

1131

        if (count > TRY_ERROR_LIMIT) break;

×

1132

        count++;

×

1135

    valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);

×

1136

    if (valid < 0) break;

×

1139

END:

×

1140

  if (pCursor->pMeta) metaULock(pCursor->pMeta);

×

1141

  if (pCursor->pCur) tdbTbcClose(pCursor->pCur);

×

1142

  taosMemoryFree(pCursor);

×

1143

  return ret;

×

1146

int32_t metaFilterTableName(void *pVnode, SMetaFltParam *arg, SArray *pUids) {

×

1147

  SMeta         *pMeta = ((SVnode *)pVnode)->pMeta;

×

1148

  SMetaFltParam *param = arg;

×

1149

  int32_t        ret = 0;

×

1150

  char          *buf = NULL;

×

1152

  STagIdxKey *pKey = NULL;

×

1153

  int32_t     nKey = 0;

×

1155

  SIdxCursor *pCursor = NULL;

×

1156

  pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));

×

1157

  if (pCursor == NULL) {

×

1158

    return terrno;

×

1160

  pCursor->pMeta = pMeta;

×

1161

  pCursor->suid = param->suid;

×

1162

  pCursor->cid = param->cid;

×

1163

  pCursor->type = param->type;

×

1165

  char *pName = param->val;

×

1167

  metaRLock(pMeta);

×

1168

  ret = tdbTbcOpen(pMeta->pNameIdx, &pCursor->pCur, NULL);

×

1169

  if (ret != 0) {

×

1170

    goto END;

×

1173

  int cmp = 0;

×

1174

  if (tdbTbcMoveTo(pCursor->pCur, pName, strlen(pName) + 1, &cmp) < 0) {

×

1175

    goto END;

×

1177

  int32_t valid = 0;

×

1178

  int32_t count = 0;

×

1180

  int32_t TRY_ERROR_LIMIT = 1;

×

1181

  do {

×

1182

    void   *pEntryKey = NULL, *pEntryVal = NULL;

×

1183

    int32_t nEntryKey = -1, nEntryVal = 0;

×

1184

    valid = tdbTbcGet(pCursor->pCur, (const void **)pEntryKey, &nEntryKey, (const void **)&pEntryVal, &nEntryVal);

×

1185

    if (valid < 0) break;

×

1187

    if (count > TRY_ERROR_LIMIT) break;

×

1189

    char *pTableKey = (char *)pEntryKey;

×

1190

    terrno = TSDB_CODE_SUCCESS;

×

1191

    cmp = (*param->filterFunc)(pTableKey, pName, pCursor->type);

×

1192

    if (terrno != TSDB_CODE_SUCCESS) {

×

1193

      ret = terrno;

×

1194

      goto END;

×

1196

    if (cmp == 0) {

×

1197

      tb_uid_t tuid = *(tb_uid_t *)pEntryVal;

×

1198

      if (taosArrayPush(pUids, &tuid) == NULL) {

×

1199

        ret = terrno;

×

1200

        goto END;

×

1203

      if (param->equal == true) {

×

1204

        if (count > TRY_ERROR_LIMIT) break;

×

1205

        count++;

×

1208

    valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);

×

1209

    if (valid < 0) {

×

1210

      break;

×

1214

END:

×

1215

  if (pCursor->pMeta) metaULock(pCursor->pMeta);

×

1216

  if (pCursor->pCur) tdbTbcClose(pCursor->pCur);

×

1217

  taosMemoryFree(buf);

×

1218

  taosMemoryFree(pKey);

×

1220

  taosMemoryFree(pCursor);

×

1222

  return ret;

×

1224

int32_t metaFilterTtl(void *pVnode, SMetaFltParam *arg, SArray *pUids) {

×

1225

  SMeta         *pMeta = ((SVnode *)pVnode)->pMeta;

×

1226

  SMetaFltParam *param = arg;

×

1227

  int32_t        ret = 0;

×

1228

  char          *buf = NULL;

×

1230

  STtlIdxKey *pKey = NULL;

×

1231

  int32_t     nKey = 0;

×

1233

  SIdxCursor *pCursor = NULL;

×

1234

  pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));

×

1235

  if (pCursor == NULL) {

×

1236

    return terrno;

×

1238

  pCursor->pMeta = pMeta;

×

1239

  pCursor->suid = param->suid;

×

1240

  pCursor->cid = param->cid;

×

1241

  pCursor->type = param->type;

×

1243

  metaRLock(pMeta);

×

1246

END:

×

1247

  if (pCursor->pMeta) metaULock(pCursor->pMeta);

×

1248

  if (pCursor->pCur) tdbTbcClose(pCursor->pCur);

×

1249

  taosMemoryFree(buf);

×

1250

  taosMemoryFree(pKey);

×

1252

  taosMemoryFree(pCursor);

×

1254

  return ret;

×

1258

int32_t metaFilterTableIds(void *pVnode, SMetaFltParam *arg, SArray *pUids) {

×

1259

  SMeta         *pMeta = ((SVnode *)pVnode)->pMeta;

×

1260

  SMetaFltParam *param = arg;

×

1262

  SMetaEntry oStbEntry = {0};

×

1263

  int32_t    code = 0;

×

1264

  char      *buf = NULL;

×

1265

  void      *pData = NULL;

×

1266

  int        nData = 0;

×

1268

  SDecoder    dc = {0};

×

1269

  STbDbKey    tbDbKey = {0};

×

1270

  STagIdxKey *pKey = NULL;

×

1271

  int32_t     nKey = 0;

×

1273

  SIdxCursor *pCursor = NULL;

×

1274

  pCursor = (SIdxCursor *)taosMemoryCalloc(1, sizeof(SIdxCursor));

×

1275

  if (!pCursor) {

×

1276

    return terrno;

×

1278

  pCursor->pMeta = pMeta;

×

1279

  pCursor->suid = param->suid;

×

1280

  pCursor->cid = param->cid;

×

1281

  pCursor->type = param->type;

×

1283

  metaRLock(pMeta);

×

1285

  TAOS_CHECK_GOTO(tdbTbGet(pMeta->pUidIdx, &param->suid, sizeof(tb_uid_t), &pData, &nData), NULL, END);

×

1287

  tbDbKey.uid = param->suid;

×

1288

  tbDbKey.version = ((SUidIdxVal *)pData)[0].version;

×

1290

  TAOS_CHECK_GOTO(tdbTbGet(pMeta->pTbDb, &tbDbKey, sizeof(tbDbKey), &pData, &nData), NULL, END);

×

1292

  tDecoderInit(&dc, pData, nData);

×

1294

  code = metaDecodeEntry(&dc, &oStbEntry);

×

1295

  if (code) {

×

1296

    tDecoderClear(&dc);

×

1297

    goto END;

×

1300

  if (oStbEntry.stbEntry.schemaTag.pSchema == NULL || oStbEntry.stbEntry.schemaTag.pSchema == NULL) {

×

1301

    TAOS_CHECK_GOTO(TSDB_CODE_INVALID_PARA, NULL, END);

×

1304

  code = TSDB_CODE_INVALID_PARA;

×

1305

  for (int i = 0; i < oStbEntry.stbEntry.schemaTag.nCols; i++) {

×

1306

    SSchema *schema = oStbEntry.stbEntry.schemaTag.pSchema + i;

×

1307

    if (schema->colId == param->cid && param->type == schema->type && (IS_IDX_ON(schema))) {

×

1308

      code = 0;

×

1310

      TAOS_CHECK_GOTO(code, NULL, END);

×

1314

  code = tdbTbcOpen(pMeta->pTagIdx, &pCursor->pCur, NULL);

×

1315

  if (code != 0) {

×

1316

    TAOS_CHECK_GOTO(terrno, NULL, END);

×

1319

  int32_t maxSize = 0;

×

1320

  int32_t nTagData = 0;

×

1321

  void   *tagData = NULL;

×

1323

  if (param->val == NULL) {

×

1324

    metaError("vgId:%d, failed to filter NULL data", TD_VID(pMeta->pVnode));

×

1325

    goto END;

×

1327

    if (IS_VAR_DATA_TYPE(param->type)) {

×

1328

      tagData = varDataVal(param->val);

×

1329

      nTagData = varDataLen(param->val);

×

1331

      if (param->type == TSDB_DATA_TYPE_NCHAR) {

×

1332

        maxSize = 4 * nTagData + 1;

×

1333

        buf = taosMemoryCalloc(1, maxSize);

×

1334

        if (buf == NULL) {

×

1335

          TAOS_CHECK_GOTO(terrno, NULL, END);

×

1338

        if (false == taosMbsToUcs4(tagData, nTagData, (TdUcs4 *)buf, maxSize, &maxSize, NULL)) {

×

1339

          TAOS_CHECK_GOTO(terrno, NULL, END);

×

1342

        tagData = buf;

×

1343

        nTagData = maxSize;

×

1346

      tagData = param->val;

×

1347

      nTagData = tDataTypes[param->type].bytes;

×

1351

  TAOS_CHECK_GOTO(metaCreateTagIdxKey(pCursor->suid, pCursor->cid, tagData, nTagData, pCursor->type,

×

1355

  int cmp = 0;

×

1356

  TAOS_CHECK_GOTO(tdbTbcMoveTo(pCursor->pCur, pKey, nKey, &cmp), 0, END);

×

1358

  int     count = 0;

×

1359

  int32_t valid = 0;

×

1360

  bool    found = false;

×

1366

  int diffCidCount = 0;

×

1367

  do {

×

1368

    void   *entryKey = NULL, *entryVal = NULL;

×

1371

    valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, (const void **)&entryVal, &nEntryVal);

×

1372

    if (valid < 0) {

×

1373

      code = valid;

×

1375

    if (count > TRY_ERROR_LIMIT) {

×

1376

      break;

×

1379

    STagIdxKey *p = entryKey;

×

1380

    if (p == NULL) break;

×

1382

    if (p->type != pCursor->type || p->suid != pCursor->suid || p->cid != pCursor->cid) {

×

1383

      if (found == true) break;  //

×

1384

      if (diffCidCount > TRY_ERROR_LIMIT) break;

×

1385

      diffCidCount++;

×

1386

      count++;

×

1387

      valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);

×

1388

      if (valid < 0) {

×

1389

        code = valid;

×

1390

        break;

×

1392

        continue;

×

1396

    terrno = TSDB_CODE_SUCCESS;

×

1397

    int32_t cmp = (*param->filterFunc)(p->data, pKey->data, pKey->type);

×

1398

    if (terrno != TSDB_CODE_SUCCESS) {

×

1399

      TAOS_CHECK_GOTO(terrno, NULL, END);

×

1400

      break;

×

1402

    if (cmp == 0) {

×

1404

      tb_uid_t tuid = 0;

×

1405

      if (IS_VAR_DATA_TYPE(pKey->type)) {

×

1406

        tuid = *(tb_uid_t *)(p->data + varDataTLen(p->data));

×

1408

        tuid = *(tb_uid_t *)(p->data + tDataTypes[pCursor->type].bytes);

×

1410

      if (taosArrayPush(pUids, &tuid) == NULL) {

×

1411

        TAOS_CHECK_GOTO(terrno, NULL, END);

×

1413

      found = true;

×

1415

      if (param->equal == true) {

×

1416

        if (count > TRY_ERROR_LIMIT) break;

×

1417

        count++;

×

1420

    valid = param->reverse ? tdbTbcMoveToPrev(pCursor->pCur) : tdbTbcMoveToNext(pCursor->pCur);

×

1421

    if (valid < 0) {

×

1422

      code = valid;

×

1423

      break;

×

1427

END:

×

1428

  if (pCursor->pMeta) metaULock(pCursor->pMeta);

×

1429

  if (pCursor->pCur) tdbTbcClose(pCursor->pCur);

×

1430

  if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf);

×

1431

  tDecoderClear(&dc);

×

1432

  tdbFree(pData);

×

1434

  taosMemoryFree(buf);

×

1435

  taosMemoryFree(pKey);

×

1437

  taosMemoryFree(pCursor);

×

1439

  return code;

×

1442

static int32_t metaGetTableTagByUid(SMeta *pMeta, int64_t suid, int64_t uid, void **tag, int32_t *len, bool lock) {

×

1443

  int ret = 0;

×

1444

  if (lock) {

×

1445

    metaRLock(pMeta);

×

1448

  SCtbIdxKey ctbIdxKey = {.suid = suid, .uid = uid};

×

1449

  ret = tdbTbGet(pMeta->pCtbIdx, &ctbIdxKey, sizeof(SCtbIdxKey), tag, len);

×

1450

  if (lock) {

×

1451

    metaULock(pMeta);

×

1454

  return ret;

×

1457

int32_t metaGetTableTagsByUids(void *pVnode, int64_t suid, SArray *uidList) {

×

1458

  SMeta        *pMeta = ((SVnode *)pVnode)->pMeta;

×

1459

  const int32_t LIMIT = 128;

×

1461

  int32_t isLock = false;

×

1462

  int32_t sz = uidList ? taosArrayGetSize(uidList) : 0;

×

1463

  for (int i = 0; i < sz; i++) {

×

1464

    STUidTagInfo *p = taosArrayGet(uidList, i);

×

1466

    if (i % LIMIT == 0) {

×

1467

      if (isLock) metaULock(pMeta);

×

1469

      metaRLock(pMeta);

×

1470

      isLock = true;

×

1474

    void   *val = NULL;

×

1475

    int32_t len = 0;

×

1476

    if (metaGetTableTagByUid(pMeta, suid, p->uid, &val, &len, false) == 0) {

×

1477

      p->pTagVal = taosMemoryMalloc(len);

×

1478

      if (!p->pTagVal) {

×

1479

        if (isLock) metaULock(pMeta);

×

1481

        TAOS_RETURN(terrno);

×

1483

      memcpy(p->pTagVal, val, len);

×

1484

      tdbFree(val);

×

1486

      metaError("vgId:%d, failed to table tags, suid: %" PRId64 ", uid: %" PRId64 "", TD_VID(pMeta->pVnode), suid,

×

1491

  if (isLock) metaULock(pMeta);

×

1492

  return 0;

×

1495

int32_t metaGetTableTags(void *pVnode, uint64_t suid, SArray *pUidTagInfo) {

×

1496

  SMCtbCursor *pCur = metaOpenCtbCursor(pVnode, suid, 1);

×

1497

  if (!pCur) {

×

1498

    TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);

×

1504

  SHashObj *pSepecifiedUidMap = NULL;

×

1505

  size_t    numOfElems = taosArrayGetSize(pUidTagInfo);

×

1506

  if (numOfElems > 0) {

×

1508

        taosHashInit(numOfElems / 0.7, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);

×

1509

    for (int i = 0; i < numOfElems; i++) {

×

1510

      STUidTagInfo *pTagInfo = taosArrayGet(pUidTagInfo, i);

×

1511

      int32_t       code = taosHashPut(pSepecifiedUidMap, &pTagInfo->uid, sizeof(uint64_t), &i, sizeof(int32_t));

×

1512

      if (code) {

×

1513

        metaCloseCtbCursor(pCur);

×

1514

        taosHashCleanup(pSepecifiedUidMap);

×

1515

        return code;

×

1520

  if (numOfElems == 0) {  // all data needs to be added into the pUidTagInfo list

×

1521

    while (1) {

×

1522

      tb_uid_t uid = metaCtbCursorNext(pCur);

×

1523

      if (uid == 0) {

×

1524

        break;

×

1527

      STUidTagInfo info = {.uid = uid, .pTagVal = pCur->pVal};

×

1528

      info.pTagVal = taosMemoryMalloc(pCur->vLen);

×

1529

      if (!info.pTagVal) {

×

1530

        metaCloseCtbCursor(pCur);

×

1531

        taosHashCleanup(pSepecifiedUidMap);

×

1532

        return terrno;

×

1534

      memcpy(info.pTagVal, pCur->pVal, pCur->vLen);

×

1535

      if (taosArrayPush(pUidTagInfo, &info) == NULL) {

×

1536

        taosMemoryFreeClear(info.pTagVal);

×

1537

        metaCloseCtbCursor(pCur);

×

1538

        taosHashCleanup(pSepecifiedUidMap);

×

1539

        return terrno;

×

1543

    while (1) {

×

1544

      tb_uid_t uid = metaCtbCursorNext(pCur);

×

1545

      if (uid == 0) {

×

1546

        break;

×

1549

      int32_t *index = taosHashGet(pSepecifiedUidMap, &uid, sizeof(uint64_t));

×

1550

      if (index == NULL) {

×

1551

        continue;

×

1554

      STUidTagInfo *pTagInfo = taosArrayGet(pUidTagInfo, *index);

×

1555

      if (pTagInfo->pTagVal == NULL) {

×

1556

        pTagInfo->pTagVal = taosMemoryMalloc(pCur->vLen);

×

1557

        if (!pTagInfo->pTagVal) {

×

1558

          metaCloseCtbCursor(pCur);

×

1559

          taosHashCleanup(pSepecifiedUidMap);

×

1560

          return terrno;

×

1562

        memcpy(pTagInfo->pTagVal, pCur->pVal, pCur->vLen);

×

1567

  taosHashCleanup(pSepecifiedUidMap);

×

1568

  metaCloseCtbCursor(pCur);

×

1569

  return TSDB_CODE_SUCCESS;

×

1574

int32_t metaGetInfo(SMeta *pMeta, int64_t uid, SMetaInfo *pInfo, SMetaReader *pReader) {

×

1575

  int32_t code = 0;

×

1576

  void   *pData = NULL;

×

1577

  int     nData = 0;

×

1578

  int     lock = 0;

×

1580

  if (pReader && !(pReader->flags & META_READER_NOLOCK)) {

×

1581

    lock = 1;

×

1584

  if (!lock) metaRLock(pMeta);

×

1587

  if (metaCacheGet(pMeta, uid, pInfo) == 0) {

×

1588

    if (!lock) metaULock(pMeta);

×

1589

    goto _exit;

×

1593

  if ((code = tdbTbGet(pMeta->pUidIdx, &uid, sizeof(uid), &pData, &nData)) < 0) {

×

1595

    if (!lock) metaULock(pMeta);

×

1596

    goto _exit;

×

1599

  if (!lock) metaULock(pMeta);

×

1601

  pInfo->uid = uid;

×

1602

  pInfo->suid = ((SUidIdxVal *)pData)->suid;

×

1603

  pInfo->version = ((SUidIdxVal *)pData)->version;

×

1604

  pInfo->skmVer = ((SUidIdxVal *)pData)->skmVer;

×

1606

  if (lock) {

×

1607

    metaULock(pReader->pMeta);

×

1611

  metaWLock(pMeta);

×

1612

  int32_t ret = metaCacheUpsert(pMeta, pInfo);

×

1613

  if (ret != 0) {

×

1614

    metaError("vgId:%d, failed to upsert cache, uid:%" PRId64, TD_VID(pMeta->pVnode), uid);

×

1616

  metaULock(pMeta);

×

1618

  if (lock) {

×

1619

    metaRLock(pReader->pMeta);

×

1622

_exit:

×

1623

  tdbFree(pData);

×

1624

  return code;

×

1627

int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables, int32_t *numOfCols) {

×

1628

  int32_t code = 0;

×

1630

  if (!numOfTables && !numOfCols) goto _exit;

×

1632

  SVnode *pVnodeObj = pVnode;

×

1633

  metaRLock(pVnodeObj->pMeta);

×

1636

  SMetaStbStats state = {0};

×

1637

  if (metaStatsCacheGet(pVnodeObj->pMeta, uid, &state) == TSDB_CODE_SUCCESS) {

×

1638

    metaULock(pVnodeObj->pMeta);

×

1639

    if (numOfTables) *numOfTables = state.ctbNum;

×

1640

    if (numOfCols) *numOfCols = state.colNum;

×

1641

    goto _exit;

×

1645

  int64_t ctbNum = 0;

×

1646

  int32_t colNum = 0;

×

1647

  code = vnodeGetCtbNum(pVnode, uid, &ctbNum);

×

1648

  if (TSDB_CODE_SUCCESS == code) {

×

1649

    code = vnodeGetStbColumnNum(pVnode, uid, &colNum);

×

1651

  metaULock(pVnodeObj->pMeta);

×

1652

  if (TSDB_CODE_SUCCESS != code) {

×

1653

    goto _exit;

×

1656

  if (numOfTables) *numOfTables = ctbNum;

×

1657

  if (numOfCols) *numOfCols = colNum;

×

1659

  state.uid = uid;

×

1660

  state.ctbNum = ctbNum;

×

1661

  state.colNum = colNum;

×

1664

  metaWLock(pVnodeObj->pMeta);

×

1666

  int32_t ret = metaStatsCacheUpsert(pVnodeObj->pMeta, &state);

×

1667

  if (ret) {

×

1668

    metaError("failed to upsert stats, uid:%" PRId64 ", ctbNum:%" PRId64 ", colNum:%d", uid, ctbNum, colNum);

×

1671

  metaULock(pVnodeObj->pMeta);

×

1673

_exit:

×

1674

  return code;

×

1677

void metaUpdateStbStats(SMeta *pMeta, int64_t uid, int64_t deltaCtb, int32_t deltaCol) {

×

1678

  SMetaStbStats stats = {0};

×

1680

  if (metaStatsCacheGet(pMeta, uid, &stats) == TSDB_CODE_SUCCESS) {

×

1681

    stats.ctbNum += deltaCtb;

×

1682

    stats.colNum += deltaCol;

×

1683

    int32_t code = metaStatsCacheUpsert(pMeta, &stats);

×

1684

    if (code) {

×

1685

      metaError("vgId:%d, failed to update stats, uid:%" PRId64 ", ctbNum:%" PRId64 ", colNum:%d",

×

1689

×

taosdata / TDengine / #3599

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

taosdata / TDengine / #3599

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