taosdata/TDengine | Build 3633 | source/dnode/mnode/impl/src/mndStb.c | Coveralls

69

int32_t mndInitStb(SMnode *pMnode) {

×

70

  SSdbTable table = {

×

80

  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_STB, mndProcessCreateStbReq);

×

81

  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_STB, mndProcessAlterStbReq);

×

82

  mndSetMsgHandle(pMnode, TDMT_MND_DROP_STB, mndProcessDropStbReq);

×

83

  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_STB_RSP, mndTransProcessRsp);

×

84

  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TTL_TABLE_RSP, mndProcessDropTtltbRsp);

×

85

  mndSetMsgHandle(pMnode, TDMT_VND_TRIM_RSP, mndProcessTrimDbRsp);

×

86

  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);

×

87

  mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);

×

88

  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);

×

89

  mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer);

×

90

  mndSetMsgHandle(pMnode, TDMT_MND_TRIM_DB_TIMER, mndProcessTrimDbTimer);

×

91

  mndSetMsgHandle(pMnode, TDMT_VND_S3MIGRATE_RSP, mndProcessS3MigrateDbRsp);

×

92

  mndSetMsgHandle(pMnode, TDMT_MND_S3MIGRATE_DB_TIMER, mndProcessS3MigrateDbTimer);

×

93

  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);

×

94

  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP, mndProcessDropStbReqFromMNode);

×

95

  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP_RSP, mndTransProcessRsp);

×

96

  mndSetMsgHandle(pMnode, TDMT_MND_DROP_TB_WITH_TSMA, mndProcessDropTbWithTsma);

×

97

  mndSetMsgHandle(pMnode, TDMT_VND_FETCH_TTL_EXPIRED_TBS_RSP, mndProcessFetchTtlExpiredTbs);

×

98

  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TABLE_RSP, mndTransProcessRsp);

×

106

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb);

×

107

  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb);

×

109

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_COL, mndRetrieveStbCol);

×

110

  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_COL, mndCancelGetNextStb);

×

112

  return sdbSetTable(pMnode->pSdb, table);

×

115

void mndCleanupStb(SMnode *pMnode) {}

×

117

SSdbRaw *mndStbActionEncode(SStbObj *pStb) {

×

118

  int32_t code = 0;

×

119

  int32_t lino = 0;

×

120

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

122

  int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags) * sizeof(SSchema) + pStb->commentLen +

×

123

                 pStb->ast1Len + pStb->ast2Len + pStb->numOfColumns * sizeof(SColCmpr) + STB_RESERVE_SIZE +

×

124

                 taosArrayGetSize(pStb->pFuncs) * TSDB_FUNC_NAME_LEN;

×

125

  SSdbRaw *pRaw = sdbAllocRaw(SDB_STB, STB_VER_NUMBER, size);

×

126

  if (pRaw == NULL) goto _OVER;

×

128

  int32_t dataPos = 0;

×

129

  SDB_SET_BINARY(pRaw, dataPos, pStb->name, TSDB_TABLE_FNAME_LEN, _OVER)

×

130

  SDB_SET_BINARY(pRaw, dataPos, pStb->db, TSDB_DB_FNAME_LEN, _OVER)

×

131

  SDB_SET_INT64(pRaw, dataPos, pStb->createdTime, _OVER)

×

132

  SDB_SET_INT64(pRaw, dataPos, pStb->updateTime, _OVER)

×

133

  SDB_SET_INT64(pRaw, dataPos, pStb->uid, _OVER)

×

134

  SDB_SET_INT64(pRaw, dataPos, pStb->dbUid, _OVER)

×

135

  SDB_SET_INT32(pRaw, dataPos, pStb->tagVer, _OVER)

×

136

  SDB_SET_INT32(pRaw, dataPos, pStb->colVer, _OVER)

×

137

  SDB_SET_INT32(pRaw, dataPos, pStb->smaVer, _OVER)

×

138

  SDB_SET_INT32(pRaw, dataPos, pStb->nextColId, _OVER)

×

139

  SDB_SET_INT64(pRaw, dataPos, pStb->maxdelay[0], _OVER)

×

140

  SDB_SET_INT64(pRaw, dataPos, pStb->maxdelay[1], _OVER)

×

141

  SDB_SET_INT64(pRaw, dataPos, pStb->watermark[0], _OVER)

×

142

  SDB_SET_INT64(pRaw, dataPos, pStb->watermark[1], _OVER)

×

143

  SDB_SET_INT32(pRaw, dataPos, pStb->ttl, _OVER)

×

144

  SDB_SET_INT32(pRaw, dataPos, pStb->numOfColumns, _OVER)

×

145

  SDB_SET_INT32(pRaw, dataPos, pStb->numOfTags, _OVER)

×

146

  SDB_SET_INT32(pRaw, dataPos, pStb->numOfFuncs, _OVER)

×

147

  SDB_SET_INT32(pRaw, dataPos, pStb->commentLen, _OVER)

×

148

  SDB_SET_INT32(pRaw, dataPos, pStb->ast1Len, _OVER)

×

149

  SDB_SET_INT32(pRaw, dataPos, pStb->ast2Len, _OVER)

×

151

  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {

×

152

    SSchema *pSchema = &pStb->pColumns[i];

×

153

    SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)

×

154

    SDB_SET_INT8(pRaw, dataPos, pSchema->flags, _OVER)

×

155

    SDB_SET_INT16(pRaw, dataPos, pSchema->colId, _OVER)

×

156

    SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, _OVER)

×

157

    SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)

×

160

  for (int32_t i = 0; i < pStb->numOfTags; ++i) {

×

161

    SSchema *pSchema = &pStb->pTags[i];

×

162

    SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)

×

163

    SDB_SET_INT8(pRaw, dataPos, pSchema->flags, _OVER)

×

164

    SDB_SET_INT16(pRaw, dataPos, pSchema->colId, _OVER)

×

165

    SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, _OVER)

×

166

    SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)

×

169

  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {

×

170

    char *func = taosArrayGet(pStb->pFuncs, i);

×

171

    SDB_SET_BINARY(pRaw, dataPos, func, TSDB_FUNC_NAME_LEN, _OVER)

×

174

  if (pStb->commentLen > 0) {

×

175

    SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)

×

178

  if (pStb->ast1Len > 0) {

×

179

    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)

×

182

  if (pStb->ast2Len > 0) {

×

183

    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)

×

186

  if (pStb->pCmpr != NULL) {

×

187

    for (int i = 0; i < pStb->numOfColumns; i++) {

×

188

      SColCmpr *p = &pStb->pCmpr[i];

×

189

      SDB_SET_INT16(pRaw, dataPos, p->id, _OVER)

×

190

      SDB_SET_INT32(pRaw, dataPos, p->alg, _OVER)

×

193

  SDB_SET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)

×

194

  SDB_SET_DATALEN(pRaw, dataPos, _OVER)

×

196

  terrno = 0;

×

198

_OVER:

×

199

  if (terrno != 0) {

×

200

    mError("stb:%s, failed to encode to raw:%p since %s", pStb->name, pRaw, terrstr());

×

201

    sdbFreeRaw(pRaw);

×

202

    return NULL;

×

205

  mTrace("stb:%s, encode to raw:%p, row:%p", pStb->name, pRaw, pStb);

×

206

  return pRaw;

×

209

static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {

×

210

  int32_t code = 0;

×

211

  int32_t lino = 0;

×

212

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

213

  SSdbRow *pRow = NULL;

×

214

  SStbObj *pStb = NULL;

×

216

  int8_t sver = 0;

×

217

  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;

×

219

  if (sver > STB_VER_NUMBER) {

×

220

    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;

×

221

    goto _OVER;

×

224

  pRow = sdbAllocRow(sizeof(SStbObj));

×

225

  if (pRow == NULL) goto _OVER;

×

227

  pStb = sdbGetRowObj(pRow);

×

228

  if (pStb == NULL) goto _OVER;

×

230

  int32_t dataPos = 0;

×

231

  SDB_GET_BINARY(pRaw, dataPos, pStb->name, TSDB_TABLE_FNAME_LEN, _OVER)

×

232

  SDB_GET_BINARY(pRaw, dataPos, pStb->db, TSDB_DB_FNAME_LEN, _OVER)

×

233

  SDB_GET_INT64(pRaw, dataPos, &pStb->createdTime, _OVER)

×

234

  SDB_GET_INT64(pRaw, dataPos, &pStb->updateTime, _OVER)

×

235

  SDB_GET_INT64(pRaw, dataPos, &pStb->uid, _OVER)

×

236

  SDB_GET_INT64(pRaw, dataPos, &pStb->dbUid, _OVER)

×

237

  SDB_GET_INT32(pRaw, dataPos, &pStb->tagVer, _OVER)

×

238

  SDB_GET_INT32(pRaw, dataPos, &pStb->colVer, _OVER)

×

239

  SDB_GET_INT32(pRaw, dataPos, &pStb->smaVer, _OVER)

×

240

  SDB_GET_INT32(pRaw, dataPos, &pStb->nextColId, _OVER)

×

241

  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[0], _OVER)

×

242

  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[1], _OVER)

×

243

  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[0], _OVER)

×

244

  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[1], _OVER)

×

245

  SDB_GET_INT32(pRaw, dataPos, &pStb->ttl, _OVER)

×

246

  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfColumns, _OVER)

×

247

  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfTags, _OVER)

×

248

  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfFuncs, _OVER)

×

249

  SDB_GET_INT32(pRaw, dataPos, &pStb->commentLen, _OVER)

×

250

  SDB_GET_INT32(pRaw, dataPos, &pStb->ast1Len, _OVER)

×

251

  SDB_GET_INT32(pRaw, dataPos, &pStb->ast2Len, _OVER)

×

253

  pStb->pColumns = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchema));

×

254

  pStb->pTags = taosMemoryCalloc(pStb->numOfTags, sizeof(SSchema));

×

255

  pStb->pFuncs = taosArrayInit(pStb->numOfFuncs, TSDB_FUNC_NAME_LEN);

×

256

  if (pStb->pColumns == NULL || pStb->pTags == NULL || pStb->pFuncs == NULL) {

×

257

    goto _OVER;

×

260

  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {

×

261

    SSchema *pSchema = &pStb->pColumns[i];

×

262

    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)

×

263

    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)

×

264

    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)

×

265

    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)

×

266

    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)

×

269

  for (int32_t i = 0; i < pStb->numOfTags; ++i) {

×

270

    SSchema *pSchema = &pStb->pTags[i];

×

271

    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)

×

272

    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)

×

273

    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)

×

274

    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)

×

275

    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)

×

278

  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {

×

279

    char funcName[TSDB_FUNC_NAME_LEN] = {0};

×

280

    SDB_GET_BINARY(pRaw, dataPos, funcName, TSDB_FUNC_NAME_LEN, _OVER)

×

281

    if (taosArrayPush(pStb->pFuncs, funcName) == NULL) goto _OVER;

×

284

  if (pStb->commentLen > 0) {

×

285

    pStb->comment = taosMemoryCalloc(pStb->commentLen + 1, 1);

×

286

    if (pStb->comment == NULL) goto _OVER;

×

287

    SDB_GET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)

×

290

  if (pStb->ast1Len > 0) {

×

291

    pStb->pAst1 = taosMemoryCalloc(pStb->ast1Len, 1);

×

292

    if (pStb->pAst1 == NULL) goto _OVER;

×

293

    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)

×

296

  if (pStb->ast2Len > 0) {

×

297

    pStb->pAst2 = taosMemoryCalloc(pStb->ast2Len, 1);

×

298

    if (pStb->pAst2 == NULL) goto _OVER;

×

299

    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)

×

302

  pStb->pCmpr = taosMemoryCalloc(pStb->numOfColumns, sizeof(SColCmpr));

×

303

  if (sver < STB_VER_NUMBER) {

×

306

    for (int i = 0; i < pStb->numOfColumns; i++) {

×

307

      SSchema  *pSchema = &pStb->pColumns[i];

×

308

      SColCmpr *pCmpr = &pStb->pCmpr[i];

×

309

      pCmpr->id = pSchema->colId;

×

310

      pCmpr->alg = createDefaultColCmprByType(pSchema->type);

×

313

    for (int i = 0; i < pStb->numOfColumns; i++) {

×

314

      SColCmpr *pCmpr = &pStb->pCmpr[i];

×

315

      SDB_GET_INT16(pRaw, dataPos, &pCmpr->id, _OVER)

×

316

      SDB_GET_INT32(pRaw, dataPos, (int32_t *)&pCmpr->alg, _OVER)  // compatiable

×

320

  SDB_GET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)

×

322

  terrno = 0;

×

324

_OVER:

×

325

  if (terrno != 0) {

×

326

    mError("stb:%s, failed to decode from raw:%p since %s", pStb == NULL ? "null" : pStb->name, pRaw, terrstr());

×

327

    if (pStb != NULL) {

×

328

      taosMemoryFreeClear(pStb->pColumns);

×

329

      taosMemoryFreeClear(pStb->pTags);

×

330

      taosMemoryFreeClear(pStb->comment);

×

331

      taosMemoryFree(pStb->pCmpr);

×

333

    taosMemoryFreeClear(pRow);

×

334

    return NULL;

×

337

  mTrace("stb:%s, decode from raw:%p, row:%p", pStb->name, pRaw, pStb);

×

338

  return pRow;

×

341

void mndFreeStb(SStbObj *pStb) {

×

342

  taosArrayDestroy(pStb->pFuncs);

×

343

  taosMemoryFreeClear(pStb->pColumns);

×

344

  taosMemoryFreeClear(pStb->pTags);

×

345

  taosMemoryFreeClear(pStb->comment);

×

346

  taosMemoryFreeClear(pStb->pAst1);

×

347

  taosMemoryFreeClear(pStb->pAst2);

×

348

  taosMemoryFreeClear(pStb->pCmpr);

×

349

×

351

static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb) {

×

352

  mTrace("stb:%s, perform insert action, row:%p", pStb->name, pStb);

×

353

  return 0;

×

356

static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {

×

357

  mTrace("stb:%s, perform delete action, row:%p", pStb->name, pStb);

×

358

  mndFreeStb(pStb);

×

359

  return 0;

×

362

static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {

×

363

  mTrace("stb:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);

×

365

  taosWLockLatch(&pOld->lock);

×

366

  int32_t numOfColumns = pOld->numOfColumns;

×

367

  if (pOld->numOfColumns < pNew->numOfColumns) {

×

368

    void *pColumns = taosMemoryMalloc(pNew->numOfColumns * sizeof(SSchema));

×

369

    if (pColumns != NULL) {

×

370

      taosMemoryFree(pOld->pColumns);

×

371

      pOld->pColumns = pColumns;

×

373

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

374

      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());

×

375

      taosWUnLockLatch(&pOld->lock);

×

379

  if (pOld->numOfTags < pNew->numOfTags) {

×

380

    void *pTags = taosMemoryMalloc(pNew->numOfTags * sizeof(SSchema));

×

381

    if (pTags != NULL) {

×

382

      taosMemoryFree(pOld->pTags);

×

383

      pOld->pTags = pTags;

×

385

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

386

      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());

×

387

      taosWUnLockLatch(&pOld->lock);

×

391

  if (pOld->commentLen < pNew->commentLen && pNew->commentLen > 0) {

×

392

    void *comment = taosMemoryMalloc(pNew->commentLen + 1);

×

393

    if (comment != NULL) {

×

394

      taosMemoryFree(pOld->comment);

×

395

      pOld->comment = comment;

×

397

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

398

      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());

×

399

      taosWUnLockLatch(&pOld->lock);

×

402

  pOld->commentLen = pNew->commentLen;

×

404

  if (pOld->ast1Len < pNew->ast1Len) {

×

405

    void *pAst1 = taosMemoryMalloc(pNew->ast1Len + 1);

×

406

    if (pAst1 != NULL) {

×

407

      taosMemoryFree(pOld->pAst1);

×

408

      pOld->pAst1 = pAst1;

×

410

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

411

      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());

×

412

      taosWUnLockLatch(&pOld->lock);

×

416

  if (pOld->ast2Len < pNew->ast2Len) {

×

417

    void *pAst2 = taosMemoryMalloc(pNew->ast2Len + 1);

×

418

    if (pAst2 != NULL) {

×

419

      taosMemoryFree(pOld->pAst2);

×

420

      pOld->pAst2 = pAst2;

×

422

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

423

      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());

×

424

      taosWUnLockLatch(&pOld->lock);

×

428

  pOld->updateTime = pNew->updateTime;

×

429

  pOld->tagVer = pNew->tagVer;

×

430

  pOld->colVer = pNew->colVer;

×

431

  pOld->smaVer = pNew->smaVer;

×

432

  pOld->nextColId = pNew->nextColId;

×

433

  pOld->ttl = pNew->ttl;

×

434

  if (pNew->numOfColumns > 0) {

×

435

    pOld->numOfColumns = pNew->numOfColumns;

×

436

    memcpy(pOld->pColumns, pNew->pColumns, pOld->numOfColumns * sizeof(SSchema));

×

438

  if (pNew->numOfTags > 0) {

×

439

    pOld->numOfTags = pNew->numOfTags;

×

440

    memcpy(pOld->pTags, pNew->pTags, pOld->numOfTags * sizeof(SSchema));

×

442

  if (pNew->commentLen > 0) {

×

443

    memcpy(pOld->comment, pNew->comment, pNew->commentLen + 1);

×

444

    pOld->commentLen = pNew->commentLen;

×

446

  if (pNew->ast1Len != 0) {

×

447

    memcpy(pOld->pAst1, pNew->pAst1, pNew->ast1Len);

×

448

    pOld->ast1Len = pNew->ast1Len;

×

450

  if (pNew->ast2Len != 0) {

×

451

    memcpy(pOld->pAst2, pNew->pAst2, pNew->ast2Len);

×

452

    pOld->ast2Len = pNew->ast2Len;

×

454

  if (numOfColumns < pNew->numOfColumns) {

×

455

    taosMemoryFree(pOld->pCmpr);

×

456

    pOld->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));

×

457

    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));

×

459

    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));

×

462

  taosWUnLockLatch(&pOld->lock);

×

463

  return 0;

×

466

SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName) {

×

467

  SSdb    *pSdb = pMnode->pSdb;

×

468

  SStbObj *pStb = sdbAcquire(pSdb, SDB_STB, stbName);

×

469

  if (pStb == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {

×

470

    terrno = TSDB_CODE_MND_STB_NOT_EXIST;

×

472

  return pStb;

×

475

void mndReleaseStb(SMnode *pMnode, SStbObj *pStb) {

×

476

  SSdb *pSdb = pMnode->pSdb;

×

477

  sdbRelease(pSdb, pStb);

×

478

×

480

SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {

×

481

  SName name = {0};

×

482

  if ((terrno = tNameFromString(&name, stbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) return NULL;

×

484

  char db[TSDB_TABLE_FNAME_LEN] = {0};

×

485

  if ((terrno = tNameGetFullDbName(&name, db)) != 0) return NULL;

×

487

  return mndAcquireDb(pMnode, db);

×

499

void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen, void *alterOriData,

×

501

  SEncoder       encoder = {0};

×

503

  SName          name = {0};

×

504

  SVCreateStbReq req = {0};

×

506

  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {

×

507

    goto _err;

×

509

  char dbFName[TSDB_DB_FNAME_LEN] = {0};

×

510

  if ((terrno = tNameGetFullDbName(&name, dbFName)) != 0) {

×

511

    goto _err;

×

514

  req.name = (char *)tNameGetTableName(&name);

×

515

  req.suid = pStb->uid;

×

516

  req.rollup = pStb->ast1Len > 0 ? 1 : 0;

×

517

  req.alterOriData = alterOriData;

×

518

  req.alterOriDataLen = alterOriDataLen;

×

519

  req.source = pStb->source;

×

521

  req.schemaRow.nCols = pStb->numOfColumns;

×

522

  req.schemaRow.version = pStb->colVer;

×

523

  req.schemaRow.pSchema = pStb->pColumns;

×

524

  req.schemaTag.nCols = pStb->numOfTags;

×

525

  req.schemaTag.version = pStb->tagVer;

×

526

  req.schemaTag.pSchema = pStb->pTags;

×

528

  req.colCmpred = 1;

×

529

  SColCmprWrapper *pCmpr = &req.colCmpr;

×

530

  pCmpr->version = pStb->colVer;

×

531

  pCmpr->nCols = pStb->numOfColumns;

×

533

  req.colCmpr.pColCmpr = taosMemoryCalloc(pCmpr->nCols, sizeof(SColCmpr));

×

534

  for (int32_t i = 0; i < pStb->numOfColumns; i++) {

×

535

    SColCmpr *p = &pCmpr->pColCmpr[i];

×

536

    p->alg = pStb->pCmpr[i].alg;

×

537

    p->id = pStb->pCmpr[i].id;

×

540

  if (req.rollup) {

×

541

    req.rsmaParam.maxdelay[0] = pStb->maxdelay[0];

×

542

    req.rsmaParam.maxdelay[1] = pStb->maxdelay[1];

×

543

    req.rsmaParam.watermark[0] = pStb->watermark[0];

×

544

    req.rsmaParam.watermark[1] = pStb->watermark[1];

×

545

    if (pStb->ast1Len > 0) {

×

546

      if (mndConvertRsmaTask(&req.rsmaParam.qmsg[0], &req.rsmaParam.qmsgLen[0], pStb->pAst1, pStb->uid,

×

549

        goto _err;

×

552

    if (pStb->ast2Len > 0) {

×

553

      if (mndConvertRsmaTask(&req.rsmaParam.qmsg[1], &req.rsmaParam.qmsgLen[1], pStb->pAst2, pStb->uid,

×

556

        goto _err;

×

561

  int32_t ret = 0;

×

562

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

×

563

  if (ret < 0) {

×

564

    goto _err;

×

567

  contLen += sizeof(SMsgHead);

×

569

  SMsgHead *pHead = taosMemoryCalloc(1, contLen);

×

570

  if (pHead == NULL) {

×

571

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

572

    goto _err;

×

575

  pHead->contLen = htonl(contLen);

×

576

  pHead->vgId = htonl(pVgroup->vgId);

×

578

  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));

×

579

  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));

×

580

  if (tEncodeSVCreateStbReq(&encoder, &req) < 0) {

×

581

    taosMemoryFreeClear(pHead);

×

582

    tEncoderClear(&encoder);

×

583

    goto _err;

×

585

  tEncoderClear(&encoder);

×

587

  *pContLen = contLen;

×

588

  taosMemoryFreeClear(req.rsmaParam.qmsg[0]);

×

589

  taosMemoryFreeClear(req.rsmaParam.qmsg[1]);

×

590

  taosMemoryFreeClear(req.colCmpr.pColCmpr);

×

591

  return pHead;

×

592

_err:

×

593

  taosMemoryFreeClear(req.rsmaParam.qmsg[0]);

×

594

  taosMemoryFreeClear(req.rsmaParam.qmsg[1]);

×

595

  taosMemoryFreeClear(req.colCmpr.pColCmpr);

×

596

  return NULL;

×

599

static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {

×

600

  SName        name = {0};

×

601

  SVDropStbReq req = {0};

×

602

  int32_t      contLen = 0;

×

603

  int32_t      ret = 0;

×

604

  SMsgHead    *pHead = NULL;

×

605

  SEncoder     encoder = {0};

×

607

  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {

×

608

    return NULL;

×

611

  req.name = (char *)tNameGetTableName(&name);

×

612

  req.suid = pStb->uid;

×

614

  tEncodeSize(tEncodeSVDropStbReq, &req, contLen, ret);

×

615

  if (ret < 0) return NULL;

×

617

  contLen += sizeof(SMsgHead);

×

618

  pHead = taosMemoryMalloc(contLen);

×

619

  if (pHead == NULL) {

×

620

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

621

    return NULL;

×

624

  pHead->contLen = htonl(contLen);

×

625

  pHead->vgId = htonl(pVgroup->vgId);

×

627

  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));

×

629

  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));

×

630

  int32_t code = tEncodeSVDropStbReq(&encoder, &req);

×

631

  tEncoderClear(&encoder);

×

632

  if (code != 0) {

×

633

    terrno = code;

×

634

    return NULL;

×

637

  *pContLen = contLen;

×

638

  return pHead;

×

641

int32_t mndCheckCreateStbReq(SMCreateStbReq *pCreate) {

×

642

  int32_t code = 0;

×

643

  if (pCreate->igExists < 0 || pCreate->igExists > 1) {

×

644

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

645

    TAOS_RETURN(code);

×

648

  if (pCreate->numOfColumns < TSDB_MIN_COLUMNS || pCreate->numOfTags + pCreate->numOfColumns > TSDB_MAX_COLUMNS) {

×

649

    code = TSDB_CODE_PAR_INVALID_COLUMNS_NUM;

×

650

    TAOS_RETURN(code);

×

653

  if (pCreate->numOfTags <= 0 || pCreate->numOfTags > TSDB_MAX_TAGS) {

×

654

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

655

    TAOS_RETURN(code);

×

658

  SField *pField = taosArrayGet(pCreate->pColumns, 0);

×

659

  if (pField->type != TSDB_DATA_TYPE_TIMESTAMP) {

×

660

    code = TSDB_CODE_PAR_INVALID_FIRST_COLUMN;

×

661

    TAOS_RETURN(code);

×

664

  for (int32_t i = 0; i < pCreate->numOfColumns; ++i) {

×

665

    SFieldWithOptions *pField1 = taosArrayGet(pCreate->pColumns, i);

×

666

    if (pField1->type >= TSDB_DATA_TYPE_MAX) {

×

667

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

668

      TAOS_RETURN(code);

×

670

    if (pField1->bytes <= 0) {

×

671

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

672

      TAOS_RETURN(code);

×

674

    if (pField1->name[0] == 0) {

×

675

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

676

      TAOS_RETURN(code);

×

680

  for (int32_t i = 0; i < pCreate->numOfTags; ++i) {

×

681

    SField *pField1 = taosArrayGet(pCreate->pTags, i);

×

682

    if (pField1->type >= TSDB_DATA_TYPE_MAX) {

×

683

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

684

      TAOS_RETURN(code);

×

686

    if (pField1->bytes <= 0) {

×

687

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

688

      TAOS_RETURN(code);

×

690

    if (pField1->name[0] == 0) {

×

691

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

692

      TAOS_RETURN(code);

×

696

  TAOS_RETURN(code);

×

699

static int32_t mndSetCreateStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

700

  int32_t  code = 0;

×

701

  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);

×

702

  if (pRedoRaw == NULL) {

×

703

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

704

    if (terrno != 0) code = terrno;

×

705

    TAOS_RETURN(code);

×

707

  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {

×

708

    sdbFreeRaw(pRedoRaw);

×

709

    TAOS_RETURN(code);

×

711

  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));

×

713

  TAOS_RETURN(code);

×

716

static int32_t mndSetCreateStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

717

  int32_t  code = 0;

×

718

  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);

×

719

  if (pCommitRaw == NULL) {

×

720

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

721

    if (terrno != 0) code = terrno;

×

722

    TAOS_RETURN(code);

×

724

  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {

×

725

    sdbFreeRaw(pCommitRaw);

×

726

    TAOS_RETURN(code);

×

728

  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

×

730

  TAOS_RETURN(code);

×

733

static int32_t mndSetCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

734

  int32_t code = 0;

×

735

  SSdb   *pSdb = pMnode->pSdb;

×

736

  SVgObj *pVgroup = NULL;

×

737

  void   *pIter = NULL;

×

740

  while (1) {

×

741

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

742

    if (pIter == NULL) break;

×

743

    if (!mndVgroupInDb(pVgroup, pDb->uid)) {

×

744

      sdbRelease(pSdb, pVgroup);

×

745

      continue;

×

748

    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);

×

749

    if (pReq == NULL) {

×

750

      sdbCancelFetch(pSdb, pIter);

×

751

      sdbRelease(pSdb, pVgroup);

×

752

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

753

      if (terrno != 0) code = terrno;

×

754

      TAOS_RETURN(code);

×

757

    STransAction action = {0};

×

758

    action.mTraceId = pTrans->mTraceId;

×

759

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

760

    action.pCont = pReq;

×

761

    action.contLen = contLen;

×

762

    action.msgType = TDMT_VND_CREATE_STB;

×

763

    action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;

×

764

    action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

765

    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {

×

766

      taosMemoryFree(pReq);

×

767

      sdbCancelFetch(pSdb, pIter);

×

768

      sdbRelease(pSdb, pVgroup);

×

769

      TAOS_RETURN(code);

×

771

    sdbRelease(pSdb, pVgroup);

×

774

  TAOS_RETURN(code);

×

777

int32_t mndSetForceDropCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, SStbObj *pStb) {

×

778

  int32_t code = 0;

×

779

  SSdb   *pSdb = pMnode->pSdb;

×

782

  void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);

×

783

  if (pReq == NULL) {

×

784

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

785

    if (terrno != 0) code = terrno;

×

786

    TAOS_RETURN(code);

×

789

  STransAction action = {0};

×

790

  action.mTraceId = pTrans->mTraceId;

×

791

  action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

792

  action.pCont = pReq;

×

793

  action.contLen = contLen;

×

794

  action.msgType = TDMT_VND_CREATE_STB;

×

795

  action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;

×

796

  action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

797

  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {

×

798

    taosMemoryFree(pReq);

×

799

    TAOS_RETURN(code);

×

802

  TAOS_RETURN(code);

×

805

static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

806

  int32_t code = 0;

×

807

  SSdb   *pSdb = pMnode->pSdb;

×

808

  SVgObj *pVgroup = NULL;

×

809

  void   *pIter = NULL;

×

811

  while (1) {

×

812

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

813

    if (pIter == NULL) break;

×

814

    if (!mndVgroupInDb(pVgroup, pDb->uid)) {

×

815

      sdbRelease(pSdb, pVgroup);

×

816

      continue;

×

819

    int32_t contLen = 0;

×

820

    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);

×

821

    if (pReq == NULL) {

×

822

      sdbCancelFetch(pSdb, pIter);

×

823

      sdbRelease(pSdb, pVgroup);

×

824

      code = TSDB_CODE_OUT_OF_MEMORY;

×

825

      TAOS_RETURN(code);

×

828

    STransAction action = {0};

×

829

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

830

    action.pCont = pReq;

×

831

    action.contLen = contLen;

×

832

    action.msgType = TDMT_VND_DROP_STB;

×

833

    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

834

    if ((code = mndTransAppendUndoAction(pTrans, &action)) != 0) {

×

835

      taosMemoryFree(pReq);

×

836

      sdbCancelFetch(pSdb, pIter);

×

837

      sdbRelease(pSdb, pVgroup);

×

838

      TAOS_RETURN(code);

×

840

    sdbRelease(pSdb, pVgroup);

×

843

  TAOS_RETURN(code);

×

846

static SSchema *mndFindStbColumns(const SStbObj *pStb, const char *colName) {

×

847

  for (int32_t col = 0; col < pStb->numOfColumns; ++col) {

×

848

    SSchema *pSchema = &pStb->pColumns[col];

×

849

    if (strncasecmp(pSchema->name, colName, TSDB_COL_NAME_LEN) == 0) {

×

850

      return pSchema;

×

853

  return NULL;

×

856

int32_t mndBuildStbFromReq(SMnode *pMnode, SStbObj *pDst, SMCreateStbReq *pCreate, SDbObj *pDb) {

×

857

  int32_t code = 0;

×

858

  memcpy(pDst->name, pCreate->name, TSDB_TABLE_FNAME_LEN);

×

859

  memcpy(pDst->db, pDb->name, TSDB_DB_FNAME_LEN);

×

860

  pDst->createdTime = taosGetTimestampMs();

×

861

  pDst->updateTime = pDst->createdTime;

×

862

  pDst->uid = (pCreate->source == TD_REQ_FROM_TAOX_OLD || pCreate->source == TD_REQ_FROM_TAOX)

×

864

                  : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);

×

865

  pDst->dbUid = pDb->uid;

×

866

  pDst->tagVer = 1;

×

867

  pDst->colVer = 1;

×

868

  pDst->smaVer = 1;

×

869

  pDst->nextColId = 1;

×

870

  pDst->maxdelay[0] = pCreate->delay1;

×

871

  pDst->maxdelay[1] = pCreate->delay2;

×

872

  pDst->watermark[0] = pCreate->watermark1;

×

873

  pDst->watermark[1] = pCreate->watermark2;

×

874

  pDst->ttl = pCreate->ttl;

×

875

  pDst->numOfColumns = pCreate->numOfColumns;

×

876

  pDst->numOfTags = pCreate->numOfTags;

×

877

  pDst->numOfFuncs = pCreate->numOfFuncs;

×

878

  pDst->commentLen = pCreate->commentLen;

×

879

  pDst->pFuncs = pCreate->pFuncs;

×

880

  pDst->source = pCreate->source;

×

881

  pCreate->pFuncs = NULL;

×

883

  if (pDst->commentLen > 0) {

×

884

    pDst->comment = taosMemoryCalloc(pDst->commentLen + 1, 1);

×

885

    if (pDst->comment == NULL) {

×

886

      code = terrno;

×

887

      TAOS_RETURN(code);

×

889

    memcpy(pDst->comment, pCreate->pComment, pDst->commentLen + 1);

×

892

  pDst->ast1Len = pCreate->ast1Len;

×

893

  if (pDst->ast1Len > 0) {

×

894

    pDst->pAst1 = taosMemoryCalloc(pDst->ast1Len, 1);

×

895

    if (pDst->pAst1 == NULL) {

×

896

      code = terrno;

×

897

      TAOS_RETURN(code);

×

899

    memcpy(pDst->pAst1, pCreate->pAst1, pDst->ast1Len);

×

902

  pDst->ast2Len = pCreate->ast2Len;

×

903

  if (pDst->ast2Len > 0) {

×

904

    pDst->pAst2 = taosMemoryCalloc(pDst->ast2Len, 1);

×

905

    if (pDst->pAst2 == NULL) {

×

906

      code = terrno;

×

907

      TAOS_RETURN(code);

×

909

    memcpy(pDst->pAst2, pCreate->pAst2, pDst->ast2Len);

×

912

  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));

×

913

  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));

×

914

  if (pDst->pColumns == NULL || pDst->pTags == NULL) {

×

915

    code = terrno;

×

916

    TAOS_RETURN(code);

×

919

  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {

×

920

    code = TSDB_CODE_OUT_OF_RANGE;

×

921

    TAOS_RETURN(code);

×

924

  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {

×

925

    SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);

×

926

    SSchema           *pSchema = &pDst->pColumns[i];

×

927

    pSchema->type = pField->type;

×

928

    pSchema->bytes = pField->bytes;

×

929

    pSchema->flags = pField->flags;

×

930

    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

931

    pSchema->colId = pDst->nextColId;

×

932

    pDst->nextColId++;

×

935

  for (int32_t i = 0; i < pDst->numOfTags; ++i) {

×

936

    SField  *pField = taosArrayGet(pCreate->pTags, i);

×

937

    SSchema *pSchema = &pDst->pTags[i];

×

938

    pSchema->type = pField->type;

×

939

    pSchema->bytes = pField->bytes;

×

940

    if (i == 0) {

×

941

      SSCHMEA_SET_IDX_ON(pSchema);

×

943

    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

944

    pSchema->colId = pDst->nextColId;

×

945

    pDst->nextColId++;

×

948

  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SCmprObj));

×

949

  for (int32_t i = 0; i < pDst->numOfColumns; i++) {

×

950

    SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);

×

951

    SSchema           *pSchema = &pDst->pColumns[i];

×

953

    SColCmpr *pColCmpr = &pDst->pCmpr[i];

×

954

    pColCmpr->id = pSchema->colId;

×

955

    pColCmpr->alg = pField->compress;

×

957

  TAOS_RETURN(code);

×

959

static int32_t mndGenIdxNameForFirstTag(char *fullname, char *dbname, char *stbname, char *tagname) {

×

960

  SName name = {0};

×

961

  if ((terrno = tNameFromString(&name, stbname, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {

×

962

    return -1;

×

964

  return snprintf(fullname, TSDB_INDEX_FNAME_LEN, "%s.%s_%s", dbname, tagname, tNameGetTableName(&name));

×

967

static int32_t mndCreateStb(SMnode *pMnode, SRpcMsg *pReq, SMCreateStbReq *pCreate, SDbObj *pDb) {

×

968

  SStbObj stbObj = {0};

×

969

  int32_t code = -1;

×

971

  char fullIdxName[TSDB_INDEX_FNAME_LEN * 2] = {0};

×

973

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb");

×

974

  if (pTrans == NULL) {

×

975

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

976

    if (terrno != 0) code = terrno;

×

977

    goto _OVER;

×

980

  mInfo("trans:%d, used to create stb:%s", pTrans->id, pCreate->name);

×

981

  TAOS_CHECK_GOTO(mndBuildStbFromReq(pMnode, &stbObj, pCreate, pDb), NULL, _OVER);

×

983

  SSchema *pSchema = &(stbObj.pTags[0]);

×

984

  if (mndGenIdxNameForFirstTag(fullIdxName, pDb->name, stbObj.name, pSchema->name) < 0) {

×

985

    goto _OVER;

×

987

  SSIdx idx = {0};

×

988

  if (mndAcquireGlobalIdx(pMnode, fullIdxName, SDB_IDX, &idx) == 0 && idx.pIdx != NULL) {

×

989

    code = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;

×

990

    mndReleaseIdx(pMnode, idx.pIdx);

×

991

    goto _OVER;

×

994

  SIdxObj idxObj = {0};

×

995

  memcpy(idxObj.name, fullIdxName, TSDB_INDEX_FNAME_LEN);

×

996

  memcpy(idxObj.stb, stbObj.name, TSDB_TABLE_FNAME_LEN);

×

997

  memcpy(idxObj.db, stbObj.db, TSDB_DB_FNAME_LEN);

×

998

  memcpy(idxObj.colName, pSchema->name, TSDB_COL_NAME_LEN);

×

999

  idxObj.createdTime = taosGetTimestampMs();

×

1000

  idxObj.uid = mndGenerateUid(fullIdxName, strlen(fullIdxName));

×

1001

  idxObj.stbUid = stbObj.uid;

×

1002

  idxObj.dbUid = stbObj.dbUid;

×

1004

  TAOS_CHECK_GOTO(mndSetCreateIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);

×

1006

  TAOS_CHECK_GOTO(mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj), NULL, _OVER);

×

1007

  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);

×

1008

  code = 0;

×

1010

_OVER:

×

1011

  mndTransDrop(pTrans);

×

1012

  if (mndStbActionDelete(pMnode->pSdb, &stbObj) != 0) mError("failed to mndStbActionDelete");

×

1013

  TAOS_RETURN(code);

×

1016

int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

1017

  mndTransSetDbName(pTrans, pDb->name, pStb->name);

×

1018

  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));

×

1019

  TAOS_CHECK_RETURN(mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, pStb));

×

1020

  TAOS_CHECK_RETURN(mndSetCreateStbRedoActions(pMnode, pTrans, pDb, pStb));

×

1021

  TAOS_CHECK_RETURN(mndSetCreateStbUndoActions(pMnode, pTrans, pDb, pStb));

×

1022

  return 0;

×

1025

static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {

×

1026

  SMnode           *pMnode = pReq->info.node;

×

1027

  SSdb             *pSdb = pMnode->pSdb;

×

1028

  SVgObj           *pVgroup = NULL;

×

1029

  void             *pIter = NULL;

×

1030

  SVDropTtlTableReq ttlReq = {

×

1031

      .timestampSec = taosGetTimestampSec(), .ttlDropMaxCount = tsTtlBatchDropNum, .nUids = 0, .pTbUids = NULL};

×

1032

  int32_t reqLen = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq);

×

1033

  int32_t contLen = reqLen + sizeof(SMsgHead);

×

1035

  mDebug("start to process ttl timer");

×

1037

  while (1) {

×

1038

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

1039

    if (pIter == NULL) break;

×

1041

    int32_t   code = 0;

×

1042

    SMsgHead *pHead = rpcMallocCont(contLen);

×

1043

    if (pHead == NULL) {

×

1044

      sdbRelease(pSdb, pVgroup);

×

1045

      continue;

×

1047

    pHead->contLen = htonl(contLen);

×

1048

    pHead->vgId = htonl(pVgroup->vgId);

×

1049

    if ((code = tSerializeSVDropTtlTableReq((char *)pHead + sizeof(SMsgHead), reqLen, &ttlReq)) < 0) {

×

1050

      mError("vgId:%d, failed to serialize drop ttl table request since %s", pVgroup->vgId, tstrerror(code));

×

1051

      sdbRelease(pSdb, pVgroup);

×

1052

      continue;

×

1055

    SRpcMsg rpcMsg = {

×

1057

    SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

1058

    code = tmsgSendReq(&epSet, &rpcMsg);

×

1059

    if (code != 0) {

×

1060

      mError("vgId:%d, failed to send drop ttl table request to vnode since 0x%x", pVgroup->vgId, code);

×

1062

      mDebug("vgId:%d, send drop ttl table request to vnode, time:%" PRId32, pVgroup->vgId, ttlReq.timestampSec);

×

1064

    sdbRelease(pSdb, pVgroup);

×

1067

  return 0;

×

1070

static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {

×

1071

  SMnode     *pMnode = pReq->info.node;

×

1072

  SSdb       *pSdb = pMnode->pSdb;

×

1073

  SVgObj     *pVgroup = NULL;

×

1074

  void       *pIter = NULL;

×

1075

  SVTrimDbReq trimReq = {.timestamp = taosGetTimestampSec()};

×

1076

  int32_t     reqLen = tSerializeSVTrimDbReq(NULL, 0, &trimReq);

×

1077

  int32_t     contLen = reqLen + sizeof(SMsgHead);

×

1079

  while (1) {

×

1080

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

1081

    if (pIter == NULL) break;

×

1083

    int32_t code = 0;

×

1085

    SMsgHead *pHead = rpcMallocCont(contLen);

×

1086

    if (pHead == NULL) {

×

1087

      sdbCancelFetch(pSdb, pVgroup);

×

1088

      sdbRelease(pSdb, pVgroup);

×

1089

      continue;

×

1091

    pHead->contLen = htonl(contLen);

×

1092

    pHead->vgId = htonl(pVgroup->vgId);

×

1093

    if ((code = tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &trimReq)) < 0) {

×

1094

      mError("vgId:%d, failed to serialize trim db request since %s", pVgroup->vgId, tstrerror(code));

×

1097

    SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM, .pCont = pHead, .contLen = contLen};

×

1098

    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

1099

    code = tmsgSendReq(&epSet, &rpcMsg);

×

1100

    if (code != 0) {

×

1101

      mError("vgId:%d, timer failed to send vnode-trim request to vnode since 0x%x", pVgroup->vgId, code);

×

1103

      mInfo("vgId:%d, timer send vnode-trim request to vnode, time:%d", pVgroup->vgId, trimReq.timestamp);

×

1105

    sdbRelease(pSdb, pVgroup);

×

1108

  return 0;

×

1111

static int32_t mndProcessS3MigrateDbTimer(SRpcMsg *pReq) {

×

1112

  SMnode          *pMnode = pReq->info.node;

×

1113

  SSdb            *pSdb = pMnode->pSdb;

×

1114

  SVgObj          *pVgroup = NULL;

×

1115

  void            *pIter = NULL;

×

1116

  SVS3MigrateDbReq s3migrateReq = {.timestamp = taosGetTimestampSec()};

×

1117

  int32_t          reqLen = tSerializeSVS3MigrateDbReq(NULL, 0, &s3migrateReq);

×

1118

  int32_t          contLen = reqLen + sizeof(SMsgHead);

×

1120

  while (1) {

×

1121

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

1122

    if (pIter == NULL) break;

×

1124

    int32_t code = 0;

×

1126

    SMsgHead *pHead = rpcMallocCont(contLen);

×

1127

    if (pHead == NULL) {

×

1128

      sdbRelease(pSdb, pVgroup);

×

1129

      continue;

×

1131

    pHead->contLen = htonl(contLen);

×

1132

    pHead->vgId = htonl(pVgroup->vgId);

×

1133

    if ((code = tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &s3migrateReq)) < 0) {

×

1134

      mError("vgId:%d, failed to serialize s3migrate db request since %s", pVgroup->vgId, tstrerror(code));

×

1135

      sdbRelease(pSdb, pVgroup);

×

1136

      continue;

×

1139

    SRpcMsg rpcMsg = {.msgType = TDMT_VND_S3MIGRATE, .pCont = pHead, .contLen = contLen};

×

1140

    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

1141

    code = tmsgSendReq(&epSet, &rpcMsg);

×

1142

    if (code != 0) {

×

1143

      mError("vgId:%d, timer failed to send vnode-s3migrate request to vnode since 0x%x", pVgroup->vgId, code);

×

1145

      mInfo("vgId:%d, timer send vnode-s3migrate request to vnode, time:%d", pVgroup->vgId, s3migrateReq.timestamp);

×

1147

    sdbRelease(pSdb, pVgroup);

×

1150

  return 0;

×

1153

static int32_t mndFindSuperTableTagIndex(const SStbObj *pStb, const char *tagName) {

×

1154

  for (int32_t tag = 0; tag < pStb->numOfTags; tag++) {

×

1155

    if (strcmp(pStb->pTags[tag].name, tagName) == 0) {

×

1156

      return tag;

×

1160

  return -1;

×

1163

static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *colName) {

×

1164

  for (int32_t col = 0; col < pStb->numOfColumns; col++) {

×

1165

    if (strcmp(pStb->pColumns[col].name, colName) == 0) {

×

1166

      return col;

×

1170

  return -1;

×

1173

static bool mndValidateSchema(SSchema *pSchemas, int32_t nSchema, SArray *pFields, int32_t maxLen) {

×

1174

  int32_t rowLen = 0;

×

1175

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

×

1176

    rowLen += (pSchemas + i)->bytes;

×

1179

  int32_t nField = taosArrayGetSize(pFields);

×

1180

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

×

1181

    rowLen += ((SField *)TARRAY_GET_ELEM(pFields, i))->bytes;

×

1184

  return rowLen <= maxLen;

×

1187

static int32_t mndBuildStbFromAlter(SStbObj *pStb, SStbObj *pDst, SMCreateStbReq *createReq) {

×

1188

  int32_t code = 0;

×

1189

  taosRLockLatch(&pStb->lock);

×

1190

  memcpy(pDst, pStb, sizeof(SStbObj));

×

1191

  taosRUnLockLatch(&pStb->lock);

×

1193

  pDst->source = createReq->source;

×

1194

  pDst->updateTime = taosGetTimestampMs();

×

1195

  pDst->numOfColumns = createReq->numOfColumns;

×

1196

  pDst->numOfTags = createReq->numOfTags;

×

1197

  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));

×

1198

  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));

×

1199

  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SColCmpr));

×

1201

  if (pDst->pColumns == NULL || pDst->pTags == NULL || pDst->pCmpr == NULL) {

×

1202

    code = terrno;

×

1203

    TAOS_RETURN(code);

×

1206

  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {

×

1207

    code = TSDB_CODE_OUT_OF_RANGE;

×

1208

    TAOS_RETURN(code);

×

1211

  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {

×

1212

    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);

×

1213

    SSchema           *pSchema = &pDst->pColumns[i];

×

1214

    pSchema->type = pField->type;

×

1215

    pSchema->bytes = pField->bytes;

×

1216

    pSchema->flags = pField->flags;

×

1217

    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

1218

    int32_t cIndex = mndFindSuperTableColumnIndex(pStb, pField->name);

×

1219

    if (cIndex >= 0) {

×

1220

      pSchema->colId = pStb->pColumns[cIndex].colId;

×

1222

      pSchema->colId = pDst->nextColId++;

×

1226

  for (int32_t i = 0; i < pDst->numOfTags; ++i) {

×

1227

    SField  *pField = taosArrayGet(createReq->pTags, i);

×

1228

    SSchema *pSchema = &pDst->pTags[i];

×

1229

    pSchema->type = pField->type;

×

1230

    pSchema->bytes = pField->bytes;

×

1231

    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

1232

    int32_t cIndex = mndFindSuperTableTagIndex(pStb, pField->name);

×

1233

    if (cIndex >= 0) {

×

1234

      pSchema->colId = pStb->pTags[cIndex].colId;

×

1236

      pSchema->colId = pDst->nextColId++;

×

1239

  for (int32_t i = 0; i < pDst->numOfColumns; i++) {

×

1240

    SColCmpr          *p = pDst->pCmpr + i;

×

1241

    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);

×

1242

    SSchema           *pSchema = &pDst->pColumns[i];

×

1243

    p->id = pSchema->colId;

×

1244

    if (pField->compress == 0) {

×

1245

      p->alg = createDefaultColCmprByType(pSchema->type);

×

1247

      p->alg = pField->compress;

×

1250

  pDst->tagVer = createReq->tagVer;

×

1251

  pDst->colVer = createReq->colVer;

×

1252

  return TSDB_CODE_SUCCESS;

×

1255

static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {

×

1256

  SMnode        *pMnode = pReq->info.node;

×

1257

  int32_t        code = -1;

×

1258

  SStbObj       *pStb = NULL;

×

1259

  SDbObj        *pDb = NULL;

×

1260

  SMCreateStbReq createReq = {0};

×

1261

  bool           isAlter = false;

×

1262

  SHashObj      *pHash = NULL;

×

1264

  if (tDeserializeSMCreateStbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {

×

1265

    code = TSDB_CODE_INVALID_MSG;

×

1266

    goto _OVER;

×

1269

  mInfo("stb:%s, start to create", createReq.name);

×

1270

  if (mndCheckCreateStbReq(&createReq) != 0) {

×

1271

    code = TSDB_CODE_INVALID_MSG;

×

1272

    goto _OVER;

×

1275

  pStb = mndAcquireStb(pMnode, createReq.name);

×

1276

  if (pStb != NULL) {

×

1277

    if (createReq.igExists) {

×

1278

      if (createReq.source == TD_REQ_FROM_APP) {

×

1279

        mInfo("stb:%s, already exist, ignore exist is set", createReq.name);

×

1280

        code = 0;

×

1281

        goto _OVER;

×

1282

      } else if (pStb->uid != createReq.suid) {

×

1283

        mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);

×

1284

        code = 0;

×

1285

        goto _OVER;

×

1286

      } else if (createReq.tagVer > 0 || createReq.colVer > 0) {

×

1287

        int32_t tagDelta = createReq.tagVer - pStb->tagVer;

×

1288

        int32_t colDelta = createReq.colVer - pStb->colVer;

×

1289

        mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d",

×

1291

        if (tagDelta <= 0 && colDelta <= 0) {

×

1292

          mInfo("stb:%s, schema version is not incremented and nothing needs to be done", createReq.name);

×

1293

          code = 0;

×

1294

          goto _OVER;

×

1295

        } else if ((tagDelta == 1 && colDelta == 0) || (tagDelta == 0 && colDelta == 1) ||

×

1296

                   (pStb->colVer == 1 && createReq.colVer > 1) || (pStb->tagVer == 1 && createReq.tagVer > 1)) {

×

1297

          isAlter = true;

×

1298

          mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);

×

1300

          mError("stb:%s, schema version increase more than 1 number, error is returned", createReq.name);

×

1301

          code = TSDB_CODE_MND_INVALID_SCHEMA_VER;

×

1302

          goto _OVER;

×

1305

        mError("stb:%s, already exist while create, input tagVer:%d colVer:%d is invalid, origin tagVer:%d colVer:%d",

×

1307

        code = TSDB_CODE_MND_INVALID_SCHEMA_VER;

×

1308

        goto _OVER;

×

1311

      code = TSDB_CODE_MND_STB_ALREADY_EXIST;

×

1312

      goto _OVER;

×

1314

  } else if (terrno != TSDB_CODE_MND_STB_NOT_EXIST) {

×

1315

    goto _OVER;

×

1316

  } else if ((createReq.source == TD_REQ_FROM_TAOX_OLD || createReq.source == TD_REQ_FROM_TAOX) &&

×

1317

             (createReq.tagVer != 1 || createReq.colVer != 1)) {

×

1318

    mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);

×

1319

    code = 0;

×

1320

    goto _OVER;

×

1323

  pHash = taosHashInit(createReq.numOfColumns + createReq.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY),

×

1325

  if (pHash == NULL) {

×

1326

    code = TSDB_CODE_OUT_OF_MEMORY;

×

1327

    goto _OVER;

×

1330

  for (int32_t i = 0; i < createReq.numOfColumns; ++i) {

×

1331

    SFieldWithOptions *pField = taosArrayGet(createReq.pColumns, i);

×

1332

    if ((code = taosHashPut(pHash, pField->name, strlen(pField->name), NULL, 0)) != 0) {

×

1333

      if (code == TSDB_CODE_DUP_KEY) {

×

1334

        code = TSDB_CODE_TSC_DUP_COL_NAMES;

×

1336

      goto _OVER;

×

1340

  for (int32_t i = 0; i < createReq.numOfTags; ++i) {

×

1341

    SField *pField = taosArrayGet(createReq.pTags, i);

×

1342

    if ((code = taosHashPut(pHash, pField->name, strlen(pField->name), NULL, 0)) != 0) {

×

1343

      if (code == TSDB_CODE_DUP_KEY) {

×

1344

        code = TSDB_CODE_TSC_DUP_COL_NAMES;

×

1346

      goto _OVER;

×

1350

  pDb = mndAcquireDbByStb(pMnode, createReq.name);

×

1351

  if (pDb == NULL) {

×

1352

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

1353

    goto _OVER;

×

1356

  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {

×

1357

    goto _OVER;

×

1360

  int32_t numOfStbs = -1;

×

1361

  if ((code = mndGetNumOfStbs(pMnode, pDb->name, &numOfStbs)) != 0) {

×

1362

    goto _OVER;

×

1365

  if (pDb->cfg.numOfStables == 1 && numOfStbs != 0) {

×

1366

    code = TSDB_CODE_MND_SINGLE_STB_MODE_DB;

×

1367

    goto _OVER;

×

1370

  if ((code = grantCheck(TSDB_GRANT_STABLE)) < 0) {

×

1371

    goto _OVER;

×

1374

  if (isAlter) {

×

1375

    bool    needRsp = false;

×

1376

    SStbObj pDst = {0};

×

1377

    if ((code = mndBuildStbFromAlter(pStb, &pDst, &createReq)) != 0) {

×

1378

      taosMemoryFreeClear(pDst.pTags);

×

1379

      taosMemoryFreeClear(pDst.pColumns);

×

1380

      taosMemoryFreeClear(pDst.pCmpr);

×

1381

      goto _OVER;

×

1384

    code = mndAlterStbImp(pMnode, pReq, pDb, &pDst, needRsp, NULL, 0);

×

1385

    taosMemoryFreeClear(pDst.pTags);

×

1386

    taosMemoryFreeClear(pDst.pColumns);

×

1387

    taosMemoryFreeClear(pDst.pCmpr);

×

1389

    code = mndCreateStb(pMnode, pReq, &createReq, pDb);

×

1391

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

1393

  SName name = {0};

×

1394

  TAOS_CHECK_RETURN(tNameFromString(&name, createReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));

×

1396

  if (createReq.sql == NULL && createReq.sqlLen == 0) {

×

1397

    char detail[1000] = {0};

×

1399

    (void)tsnprintf(detail, sizeof(detail), "dbname:%s, stable name:%s", name.dbname, name.tname);

×

1401

    auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, detail, strlen(detail));

×

1403

    auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, createReq.sql, createReq.sqlLen);

×

1405

_OVER:

×

1406

  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

×

1407

    mError("stb:%s, failed to create since %s", createReq.name, tstrerror(code));

×

1410

  mndReleaseStb(pMnode, pStb);

×

1411

  mndReleaseDb(pMnode, pDb);

×

1412

  tFreeSMCreateStbReq(&createReq);

×

1414

  if (pHash != NULL) {

×

1415

    taosHashCleanup(pHash);

×

1418

  TAOS_RETURN(code);

×

1421

static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) {

×

1422

  int32_t code = 0;

×

1423

  if (pAlter->commentLen >= 0) return 0;

×

1424

  if (pAlter->ttl != 0) return 0;

×

1426

  if (pAlter->numOfFields < 1 || pAlter->numOfFields != (int32_t)taosArrayGetSize(pAlter->pFields)) {

×

1427

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1428

    TAOS_RETURN(code);

×

1431

  for (int32_t i = 0; i < pAlter->numOfFields; ++i) {

×

1432

    SField *pField = taosArrayGet(pAlter->pFields, i);

×

1433

    if (pField->name[0] == 0) {

×

1434

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1435

      TAOS_RETURN(code);

×

1439

  TAOS_RETURN(code);

×

1442

int32_t mndAllocStbSchemas(const SStbObj *pOld, SStbObj *pNew) {

×

1443

  pNew->pTags = taosMemoryCalloc(pNew->numOfTags, sizeof(SSchema));

×

1444

  pNew->pColumns = taosMemoryCalloc(pNew->numOfColumns, sizeof(SSchema));

×

1445

  pNew->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));

×

1446

  if (pNew->pTags == NULL || pNew->pColumns == NULL || pNew->pCmpr == NULL) {

×

1447

    TAOS_RETURN(terrno);

×

1450

  memcpy(pNew->pColumns, pOld->pColumns, sizeof(SSchema) * pOld->numOfColumns);

×

1451

  memcpy(pNew->pTags, pOld->pTags, sizeof(SSchema) * pOld->numOfTags);

×

1452

  memcpy(pNew->pCmpr, pOld->pCmpr, sizeof(SColCmpr) * pOld->numOfColumns);

×

1454

  TAOS_RETURN(0);

×

1457

static int32_t mndUpdateStbCommentAndTTL(const SStbObj *pOld, SStbObj *pNew, char *pComment, int32_t commentLen,

×

1459

  int32_t code = 0;

×

1460

  if (commentLen > 0) {

×

1461

    pNew->commentLen = commentLen;

×

1462

    pNew->comment = taosMemoryCalloc(1, commentLen + 1);

×

1463

    if (pNew->comment == NULL) {

×

1464

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1465

      return -1;

×

1467

    memcpy(pNew->comment, pComment, commentLen + 1);

×

1468

  } else if (commentLen == 0) {

×

1469

    pNew->commentLen = 0;

×

1473

  if (ttl >= 0) {

×

1474

    pNew->ttl = ttl;

×

1477

  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {

×

1478

    TAOS_RETURN(code);

×

1480

  TAOS_RETURN(code);

×

1483

static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *pFields, int32_t ntags) {

×

1484

  int32_t code = 0;

×

1485

  if (pOld->numOfTags + ntags > TSDB_MAX_TAGS) {

×

1486

    code = TSDB_CODE_MND_TOO_MANY_TAGS;

×

1487

    TAOS_RETURN(code);

×

1490

  if (pOld->numOfColumns + ntags + pOld->numOfTags > TSDB_MAX_COLUMNS) {

×

1491

    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;

×

1492

    TAOS_RETURN(code);

×

1495

  if (!mndValidateSchema(pOld->pTags, pOld->numOfTags, pFields, TSDB_MAX_TAGS_LEN)) {

×

1496

    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;

×

1497

    TAOS_RETURN(code);

×

1500

  pNew->numOfTags = pNew->numOfTags + ntags;

×

1501

  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {

×

1502

    TAOS_RETURN(code);

×

1505

  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ntags) {

×

1506

    code = TSDB_CODE_OUT_OF_RANGE;

×

1507

    TAOS_RETURN(code);

×

1510

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

×

1511

    SField *pField = taosArrayGet(pFields, i);

×

1512

    if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {

×

1513

      code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1514

      TAOS_RETURN(code);

×

1517

    if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {

×

1518

      code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1519

      TAOS_RETURN(code);

×

1522

    SSchema *pSchema = &pNew->pTags[pOld->numOfTags + i];

×

1523

    pSchema->bytes = pField->bytes;

×

1524

    pSchema->type = pField->type;

×

1525

    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

1526

    pSchema->colId = pNew->nextColId;

×

1527

    pNew->nextColId++;

×

1529

    mInfo("stb:%s, start to add tag %s", pNew->name, pSchema->name);

×

1532

  pNew->tagVer++;

×

1533

  TAOS_RETURN(code);

×

1536

static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {

×

1537

  int32_t code = 0;

×

1538

  SSdb   *pSdb = pMnode->pSdb;

×

1539

  void   *pIter = NULL;

×

1540

  while (1) {

×

1541

    SMqTopicObj *pTopic = NULL;

×

1542

    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);

×

1543

    if (pIter == NULL) break;

×

1545

    mInfo("topic:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, subType:%d sql:%s",

×

1547

    if (pTopic->ast == NULL) {

×

1548

      sdbRelease(pSdb, pTopic);

×

1549

      continue;

×

1552

    SNode *pAst = NULL;

×

1553

    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {

×

1554

      code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;

×

1555

      mError("topic:%s, create ast error", pTopic->name);

×

1556

      sdbRelease(pSdb, pTopic);

×

1557

      sdbCancelFetch(pSdb, pIter);

×

1558

      TAOS_RETURN(code);

×

1561

    SNodeList *pNodeList = NULL;

×

1562

    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=

×

1564

      sdbRelease(pSdb, pTopic);

×

1565

      sdbCancelFetch(pSdb, pIter);

×

1566

      TAOS_RETURN(code);

×

1568

    SNode *pNode = NULL;

×

1569

    FOREACH(pNode, pNodeList) {

×

1570

      SColumnNode *pCol = (SColumnNode *)pNode;

×

1571

      mInfo("topic:%s, check colId:%d tableId:%" PRId64 " ctbStbUid:%" PRId64, pTopic->name, pCol->colId, pCol->tableId,

×

1574

      if (pCol->tableId != suid && pTopic->ctbStbUid != suid) {

×

1575

        mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);

×

1576

        goto NEXT;

×

1578

      if (pCol->colId > 0 && pCol->colId == colId) {

×

1579

        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;

×

1580

        mError("topic:%s, check colId:%d conflicted", pTopic->name, pCol->colId);

×

1581

        nodesDestroyNode(pAst);

×

1582

        nodesDestroyList(pNodeList);

×

1583

        sdbCancelFetch(pSdb, pIter);

×

1584

        sdbRelease(pSdb, pTopic);

×

1585

        TAOS_RETURN(code);

×

1587

      mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);

×

1590

  NEXT:

×

1591

    sdbRelease(pSdb, pTopic);

×

1592

    nodesDestroyNode(pAst);

×

1593

    nodesDestroyList(pNodeList);

×

1595

  TAOS_RETURN(code);

×

1598

static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {

×

1599

  int32_t code = 0;

×

1600

  SSdb   *pSdb = pMnode->pSdb;

×

1601

  void   *pIter = NULL;

×

1602

  while (1) {

×

1603

    SStreamObj *pStream = NULL;

×

1604

    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);

×

1605

    if (pIter == NULL) break;

×

1607

    SNode *pAst = NULL;

×

1608

    if (nodesStringToNode(pStream->ast, &pAst) != 0) {

×

1609

      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;

×

1610

      mError("stream:%s, create ast error", pStream->name);

×

1611

      sdbRelease(pSdb, pStream);

×

1612

      sdbCancelFetch(pSdb, pIter);

×

1613

      TAOS_RETURN(code);

×

1616

    SNodeList *pNodeList = NULL;

×

1617

    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=

×

1619

      sdbRelease(pSdb, pStream);

×

1620

      sdbCancelFetch(pSdb, pIter);

×

1621

      TAOS_RETURN(code);

×

1624

    SNode *pNode = NULL;

×

1625

    FOREACH(pNode, pNodeList) {

×

1626

      SColumnNode *pCol = (SColumnNode *)pNode;

×

1628

      if (pCol->tableId != suid) {

×

1629

        mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);

×

1630

        goto NEXT;

×

1632

      if (pCol->colId > 0 && pCol->colId == colId) {

×

1633

        code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;

×

1634

        mError("stream:%s, check colId:%d conflicted", pStream->name, pCol->colId);

×

1635

        nodesDestroyNode(pAst);

×

1636

        nodesDestroyList(pNodeList);

×

1637

        sdbRelease(pSdb, pStream);

×

1638

        sdbCancelFetch(pSdb, pIter);

×

1639

        TAOS_RETURN(code);

×

1641

      mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);

×

1644

  NEXT:

×

1645

    sdbRelease(pSdb, pStream);

×

1646

    nodesDestroyNode(pAst);

×

1647

    nodesDestroyList(pNodeList);

×

1649

  TAOS_RETURN(code);

×

1652

static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {

×

1653

  int32_t code = 0;

×

1654

  SSdb   *pSdb = pMnode->pSdb;

×

1655

  void   *pIter = NULL;

×

1656

  while (1) {

×

1657

    SSmaObj *pSma = NULL;

×

1658

    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);

×

1659

    if (pIter == NULL) break;

×

1661

    mInfo("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, sql:%s", pSma->name, stbFullName,

×

1664

    SNode *pAst = NULL;

×

1665

    if (nodesStringToNode(pSma->ast, &pAst) != 0) {

×

1666

      code = TSDB_CODE_SDB_INVALID_DATA_CONTENT;

×

1667

      mError("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d failed since parse AST err",

×

1669

      sdbCancelFetch(pSdb, pIter);

×

1670

      TAOS_RETURN(code);

×

1673

    SNodeList *pNodeList = NULL;

×

1674

    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=

×

1676

      sdbCancelFetch(pSdb, pIter);

×

1677

      TAOS_RETURN(code);

×

1679

    SNode *pNode = NULL;

×

1680

    FOREACH(pNode, pNodeList) {

×

1681

      SColumnNode *pCol = (SColumnNode *)pNode;

×

1682

      mInfo("tsma:%s, check colId:%d tableId:%" PRId64, pSma->name, pCol->colId, pCol->tableId);

×

1684

      if ((pCol->tableId != suid) && (pSma->stbUid != suid)) {

×

1685

        mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);

×

1686

        goto NEXT;

×

1688

      if ((pCol->colId) > 0 && (pCol->colId == colId)) {

×

1689

        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;

×

1690

        mError("tsma:%s, check colId:%d conflicted", pSma->name, pCol->colId);

×

1691

        nodesDestroyNode(pAst);

×

1692

        nodesDestroyList(pNodeList);

×

1693

        sdbRelease(pSdb, pSma);

×

1694

        sdbCancelFetch(pSdb, pIter);

×

1695

        TAOS_RETURN(code);

×

1697

      mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);

×

1700

  NEXT:

×

1701

    sdbRelease(pSdb, pSma);

×

1702

    nodesDestroyNode(pAst);

×

1703

    nodesDestroyList(pNodeList);

×

1705

  TAOS_RETURN(code);

×

1708

int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {

×

1709

  TAOS_CHECK_RETURN(mndCheckAlterColForTopic(pMnode, stbFullName, suid, colId));

×

1710

  TAOS_CHECK_RETURN(mndCheckAlterColForStream(pMnode, stbFullName, suid, colId));

×

1711

  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, stbFullName, suid, colId));

×

1712

  TAOS_RETURN(0);

×

1715

static int32_t mndDropSuperTableTag(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *tagName) {

×

1716

  int32_t code = 0;

×

1717

  int32_t tag = mndFindSuperTableTagIndex(pOld, tagName);

×

1718

  if (tag < 0) {

×

1719

    code = TSDB_CODE_MND_TAG_NOT_EXIST;

×

1720

    TAOS_RETURN(code);

×

1723

  col_id_t colId = pOld->pTags[tag].colId;

×

1724

  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));

×

1726

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1728

  memmove(pNew->pTags + tag, pNew->pTags + tag + 1, sizeof(SSchema) * (pNew->numOfTags - tag - 1));

×

1729

  pNew->numOfTags--;

×

1731

  pNew->tagVer++;

×

1736

  mInfo("stb:%s, start to drop tag %s", pNew->name, tagName);

×

1737

  TAOS_RETURN(code);

×

1740

static int32_t mndAlterStbTagName(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pFields) {

×

1741

  int32_t code = 0;

×

1742

  if ((int32_t)taosArrayGetSize(pFields) != 2) {

×

1743

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1744

    TAOS_RETURN(code);

×

1747

  SField *pField0 = taosArrayGet(pFields, 0);

×

1748

  SField *pField1 = taosArrayGet(pFields, 1);

×

1750

  const char *oldTagName = pField0->name;

×

1751

  const char *newTagName = pField1->name;

×

1753

  int32_t tag = mndFindSuperTableTagIndex(pOld, oldTagName);

×

1754

  if (tag < 0) {

×

1755

    code = TSDB_CODE_MND_TAG_NOT_EXIST;

×

1756

    TAOS_RETURN(code);

×

1759

  col_id_t colId = pOld->pTags[tag].colId;

×

1760

  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));

×

1762

  if (mndFindSuperTableTagIndex(pOld, newTagName) >= 0) {

×

1763

    code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1764

    TAOS_RETURN(code);

×

1767

  if (mndFindSuperTableColumnIndex(pOld, newTagName) >= 0) {

×

1768

    code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1769

    TAOS_RETURN(code);

×

1772

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1774

  SSchema *pSchema = (SSchema *)(pNew->pTags + tag);

×

1775

  memcpy(pSchema->name, newTagName, TSDB_COL_NAME_LEN);

×

1777

  pNew->tagVer++;

×

1778

  mInfo("stb:%s, start to modify tag %s to %s", pNew->name, oldTagName, newTagName);

×

1779

  TAOS_RETURN(code);

×

1782

static int32_t mndAlterStbTagBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {

×

1783

  int32_t code = 0;

×

1784

  int32_t tag = mndFindSuperTableTagIndex(pOld, pField->name);

×

1785

  if (tag < 0) {

×

1786

    code = TSDB_CODE_MND_TAG_NOT_EXIST;

×

1787

    TAOS_RETURN(code);

×

1790

  col_id_t colId = pOld->pTags[tag].colId;

×

1791

  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));

×

1793

  uint32_t nLen = 0;

×

1794

  for (int32_t i = 0; i < pOld->numOfTags; ++i) {

×

1795

    nLen += (pOld->pTags[i].colId == colId) ? pField->bytes : pOld->pTags[i].bytes;

×

1798

  if (nLen > TSDB_MAX_TAGS_LEN) {

×

1799

    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;

×

1800

    TAOS_RETURN(code);

×

1803

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1805

  SSchema *pTag = pNew->pTags + tag;

×

1807

  if (!(pTag->type == TSDB_DATA_TYPE_BINARY || pTag->type == TSDB_DATA_TYPE_VARBINARY ||

×

1808

        pTag->type == TSDB_DATA_TYPE_NCHAR || pTag->type == TSDB_DATA_TYPE_GEOMETRY)) {

×

1809

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1810

    TAOS_RETURN(code);

×

1813

  if (pField->bytes <= pTag->bytes) {

×

1814

    code = TSDB_CODE_MND_INVALID_ROW_BYTES;

×

1815

    TAOS_RETURN(code);

×

1818

  pTag->bytes = pField->bytes;

×

1819

  pNew->tagVer++;

×

1821

  mInfo("stb:%s, start to modify tag len %s to %d", pNew->name, pField->name, pField->bytes);

×

1822

  TAOS_RETURN(code);

×

1825

static int32_t mndUpdateSuperTableColumnCompress(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pField,

×

1829

  if (taosArrayGetSize(pField) != nCols) return TSDB_CODE_FAILED;

×

1830

  TAOS_FIELD *p = taosArrayGet(pField, 0);

×

1832

  int32_t code = 0;

×

1833

  int32_t idx = mndFindSuperTableColumnIndex(pOld, p->name);

×

1834

  if (idx == -1) {

×

1835

    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;

×

1836

    TAOS_RETURN(code);

×

1838

  SSchema *pTarget = &pOld->pColumns[idx];

×

1839

  col_id_t colId = pTarget->colId;

×

1841

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1842

  code = validColCmprByType(pTarget->type, p->bytes);

×

1843

  if (code != TSDB_CODE_SUCCESS) {

×

1844

    TAOS_RETURN(code);

×

1847

  int8_t updated = 0;

×

1848

  for (int i = 0; i < pNew->numOfColumns; i++) {

×

1849

    SColCmpr *pCmpr = &pNew->pCmpr[i];

×

1850

    if (pCmpr->id == colId) {

×

1851

      uint32_t dst = 0;

×

1852

      updated = tUpdateCompress(pCmpr->alg, p->bytes, TSDB_COLVAL_COMPRESS_DISABLED, TSDB_COLVAL_LEVEL_DISABLED,

×

1854

      if (updated > 0) pCmpr->alg = dst;

×

1855

      break;

×

1859

  if (updated == 0) {

×

1860

    code = TSDB_CODE_MND_COLUMN_COMPRESS_ALREADY_EXIST;

×

1861

    TAOS_RETURN(code);

×

1862

  } else if (updated == -1) {

×

1863

    code = TSDB_CODE_TSC_COMPRESS_LEVEL_ERROR;

×

1864

    TAOS_RETURN(code);

×

1867

  pNew->colVer++;

×

1869

  TAOS_RETURN(code);

×

1871

static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, SArray *pFields, int32_t ncols,

×

1873

  int32_t code = 0;

×

1874

  if (pOld->numOfColumns + ncols + pOld->numOfTags > TSDB_MAX_COLUMNS) {

×

1875

    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;

×

1876

    TAOS_RETURN(code);

×

1879

  if ((code = grantCheck(TSDB_GRANT_TIMESERIES)) != 0) {

×

1880

    TAOS_RETURN(code);

×

1883

  if (!mndValidateSchema(pOld->pColumns, pOld->numOfColumns, pFields, TSDB_MAX_BYTES_PER_ROW)) {

×

1884

    code = TSDB_CODE_PAR_INVALID_ROW_LENGTH;

×

1885

    TAOS_RETURN(code);

×

1888

  pNew->numOfColumns = pNew->numOfColumns + ncols;

×

1890

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1892

  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ncols) {

×

1893

    code = TSDB_CODE_OUT_OF_RANGE;

×

1894

    TAOS_RETURN(code);

×

1897

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

×

1898

    if (withCompress) {

×

1899

      SFieldWithOptions *pField = taosArrayGet(pFields, i);

×

1900

      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {

×

1901

        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1902

        TAOS_RETURN(code);

×

1905

      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {

×

1906

        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1907

        TAOS_RETURN(code);

×

1910

      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];

×

1911

      pSchema->bytes = pField->bytes;

×

1912

      pSchema->type = pField->type;

×

1913

      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

1914

      pSchema->colId = pNew->nextColId;

×

1915

      pNew->nextColId++;

×

1917

      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];

×

1918

      pCmpr->id = pSchema->colId;

×

1919

      pCmpr->alg = pField->compress;

×

1920

      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);

×

1922

      SField *pField = taosArrayGet(pFields, i);

×

1923

      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {

×

1924

        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1925

        TAOS_RETURN(code);

×

1928

      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {

×

1929

        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1930

        TAOS_RETURN(code);

×

1933

      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];

×

1934

      pSchema->bytes = pField->bytes;

×

1935

      pSchema->type = pField->type;

×

1936

      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);

×

1937

      pSchema->colId = pNew->nextColId;

×

1938

      pNew->nextColId++;

×

1940

      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];

×

1941

      pCmpr->id = pSchema->colId;

×

1942

      pCmpr->alg = createDefaultColCmprByType(pSchema->type);

×

1943

      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);

×

1947

  pNew->colVer++;

×

1948

  TAOS_RETURN(code);

×

1951

static int32_t mndDropSuperTableColumn(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *colName) {

×

1952

  int32_t code = 0;

×

1953

  int32_t col = mndFindSuperTableColumnIndex(pOld, colName);

×

1954

  if (col < 0) {

×

1955

    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;

×

1956

    TAOS_RETURN(code);

×

1959

  if (col == 0) {

×

1960

    code = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;

×

1961

    TAOS_RETURN(code);

×

1964

  if (pOld->numOfColumns == 2) {

×

1965

    code = TSDB_CODE_PAR_INVALID_DROP_COL;

×

1966

    TAOS_RETURN(code);

×

1969

  col_id_t colId = pOld->pColumns[col].colId;

×

1970

  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));

×

1972

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1974

  int32_t sz = pNew->numOfColumns - col - 1;

×

1975

  memmove(pNew->pColumns + col, pNew->pColumns + col + 1, sizeof(SSchema) * sz);

×

1976

  memmove(pNew->pCmpr + col, pNew->pCmpr + col + 1, sizeof(SColCmpr) * sz);

×

1977

  pNew->numOfColumns--;

×

1979

  pNew->colVer++;

×

1980

  mInfo("stb:%s, start to drop col %s", pNew->name, colName);

×

1981

  TAOS_RETURN(code);

×

1984

static int32_t mndAlterStbColumnBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {

×

1985

  int32_t code = 0;

×

1986

  int32_t col = mndFindSuperTableColumnIndex(pOld, pField->name);

×

1987

  if (col < 0) {

×

1988

    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;

×

1989

    TAOS_RETURN(code);

×

1992

  col_id_t colId = pOld->pColumns[col].colId;

×

1994

  uint32_t nLen = 0;

×

1995

  for (int32_t i = 0; i < pOld->numOfColumns; ++i) {

×

1996

    nLen += (pOld->pColumns[i].colId == colId) ? pField->bytes : pOld->pColumns[i].bytes;

×

1999

  if (nLen > TSDB_MAX_BYTES_PER_ROW) {

×

2000

    code = TSDB_CODE_MND_INVALID_ROW_BYTES;

×

2001

    TAOS_RETURN(code);

×

2004

  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));

×

2006

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

2008

  SSchema *pCol = pNew->pColumns + col;

×

2009

  if (!(pCol->type == TSDB_DATA_TYPE_BINARY || pCol->type == TSDB_DATA_TYPE_VARBINARY ||

×

2010

        pCol->type == TSDB_DATA_TYPE_NCHAR || pCol->type == TSDB_DATA_TYPE_GEOMETRY)) {

×

2011

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

2012

    TAOS_RETURN(code);

×

2015

  if (pField->bytes <= pCol->bytes) {

×

2016

    code = TSDB_CODE_MND_INVALID_ROW_BYTES;

×

2017

    TAOS_RETURN(code);

×

2020

  pCol->bytes = pField->bytes;

×

2021

  pNew->colVer++;

×

2023

  mInfo("stb:%s, start to modify col len %s to %d", pNew->name, pField->name, pField->bytes);

×

2024

  TAOS_RETURN(code);

×

2027

static int32_t mndSetAlterStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

2028

  int32_t  code = 0;

×

2029

  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);

×

2030

  if (pRedoRaw == NULL) {

×

2031

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2032

    if (terrno != 0) code = terrno;

×

2033

    TAOS_RETURN(code);

×

2035

  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {

×

2036

    sdbFreeRaw(pRedoRaw);

×

2037

    TAOS_RETURN(code);

×

2039

  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY));

×

2041

  TAOS_RETURN(code);

×

2044

static int32_t mndSetAlterStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

2045

  int32_t  code = 0;

×

2046

  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);

×

2047

  if (pCommitRaw == NULL) {

×

2048

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2049

    if (terrno != 0) code = terrno;

×

2050

    TAOS_RETURN(code);

×

2052

  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {

×

2053

    sdbFreeRaw(pCommitRaw);

×

2054

    TAOS_RETURN(code);

×

2056

  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

×

2058

  TAOS_RETURN(code);

×

2061

static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb, void *alterOriData,

×

2063

  int32_t code = 0;

×

2064

  SSdb   *pSdb = pMnode->pSdb;

×

2065

  SVgObj *pVgroup = NULL;

×

2066

  void   *pIter = NULL;

×

2069

  while (1) {

×

2070

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

2071

    if (pIter == NULL) break;

×

2072

    if (!mndVgroupInDb(pVgroup, pDb->uid)) {

×

2073

      sdbRelease(pSdb, pVgroup);

×

2074

      continue;

×

2077

    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);

×

2078

    if (pReq == NULL) {

×

2079

      sdbCancelFetch(pSdb, pIter);

×

2080

      sdbRelease(pSdb, pVgroup);

×

2081

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2082

      if (terrno != 0) code = terrno;

×

2083

      TAOS_RETURN(code);

×

2085

    STransAction action = {0};

×

2086

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

2087

    action.pCont = pReq;

×

2088

    action.contLen = contLen;

×

2089

    action.msgType = TDMT_VND_ALTER_STB;

×

2090

    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {

×

2091

      taosMemoryFree(pReq);

×

2092

      sdbCancelFetch(pSdb, pIter);

×

2093

      sdbRelease(pSdb, pVgroup);

×

2094

      TAOS_RETURN(code);

×

2096

    sdbRelease(pSdb, pVgroup);

×

2099

  TAOS_RETURN(code);

×

2102

static int32_t mndSetAlterStbRedoActions2(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb,

×

2104

  int32_t code = 0;

×

2105

  SSdb   *pSdb = pMnode->pSdb;

×

2106

  SVgObj *pVgroup = NULL;

×

2107

  void   *pIter = NULL;

×

2110

  while (1) {

×

2111

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

2112

    if (pIter == NULL) break;

×

2113

    if (!mndVgroupInDb(pVgroup, pDb->uid)) {

×

2114

      sdbRelease(pSdb, pVgroup);

×

2115

      continue;

×

2118

    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);

×

2119

    if (pReq == NULL) {

×

2120

      sdbCancelFetch(pSdb, pIter);

×

2121

      sdbRelease(pSdb, pVgroup);

×

2122

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2123

      if (terrno != 0) code = terrno;

×

2124

      TAOS_RETURN(code);

×

2126

    STransAction action = {0};

×

2127

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

2128

    action.pCont = pReq;

×

2129

    action.contLen = contLen;

×

2130

    action.msgType = TDMT_VND_CREATE_INDEX;

×

2131

    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {

×

2132

      taosMemoryFree(pReq);

×

2133

      sdbCancelFetch(pSdb, pIter);

×

2134

      sdbRelease(pSdb, pVgroup);

×

2135

      TAOS_RETURN(code);

×

2137

    sdbRelease(pSdb, pVgroup);

×

2140

  TAOS_RETURN(code);

×

2142

static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableMetaRsp *pRsp) {

×

2143

  int32_t code = 0;

×

2144

  taosRLockLatch(&pStb->lock);

×

2146

  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;

×

2147

  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));

×

2148

  if (pRsp->pSchemas == NULL) {

×

2149

    taosRUnLockLatch(&pStb->lock);

×

2150

    code = terrno;

×

2151

    TAOS_RETURN(code);

×

2153

  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));

×

2154

  if (pRsp->pSchemaExt == NULL) {

×

2155

    taosRUnLockLatch(&pStb->lock);

×

2156

    code = terrno;

×

2157

    TAOS_RETURN(code);

×

2160

  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));

×

2161

  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));

×

2162

  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));

×

2163

  pRsp->dbId = pDb->uid;

×

2164

  pRsp->numOfTags = pStb->numOfTags;

×

2165

  pRsp->numOfColumns = pStb->numOfColumns;

×

2166

  pRsp->precision = pDb->cfg.precision;

×

2167

  pRsp->tableType = TSDB_SUPER_TABLE;

×

2168

  pRsp->sversion = pStb->colVer;

×

2169

  pRsp->tversion = pStb->tagVer;

×

2170

  pRsp->suid = pStb->uid;

×

2171

  pRsp->tuid = pStb->uid;

×

2173

  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {

×

2174

    SSchema *pSchema = &pRsp->pSchemas[i];

×

2175

    SSchema *pSrcSchema = &pStb->pColumns[i];

×

2176

    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);

×

2177

    pSchema->type = pSrcSchema->type;

×

2178

    pSchema->flags = pSrcSchema->flags;

×

2179

    pSchema->colId = pSrcSchema->colId;

×

2180

    pSchema->bytes = pSrcSchema->bytes;

×

2183

  for (int32_t i = 0; i < pStb->numOfTags; ++i) {

×

2184

    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];

×

2185

    SSchema *pSrcSchema = &pStb->pTags[i];

×

2186

    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);

×

2187

    pSchema->type = pSrcSchema->type;

×

2188

    pSchema->flags = pSrcSchema->flags;

×

2189

    pSchema->colId = pSrcSchema->colId;

×

2190

    pSchema->bytes = pSrcSchema->bytes;

×

2192

  for (int32_t i = 0; i < pStb->numOfColumns; i++) {

×

2193

    SColCmpr   *pCmpr = &pStb->pCmpr[i];

×

2194

    SSchemaExt *pSchEx = &pRsp->pSchemaExt[i];

×

2195

    pSchEx->colId = pCmpr->id;

×

2196

    pSchEx->compress = pCmpr->alg;

×

2199

  taosRUnLockLatch(&pStb->lock);

×

2200

  TAOS_RETURN(code);

×

2203

static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableCfgRsp *pRsp) {

×

2204

  int32_t code = 0;

×

2205

  taosRLockLatch(&pStb->lock);

×

2207

  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;

×

2208

  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));

×

2209

  if (pRsp->pSchemas == NULL) {

×

2210

    taosRUnLockLatch(&pStb->lock);

×

2211

    code = terrno;

×

2212

    TAOS_RETURN(code);

×

2215

  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));

×

2216

  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));

×

2217

  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));

×

2218

  pRsp->numOfTags = pStb->numOfTags;

×

2219

  pRsp->numOfColumns = pStb->numOfColumns;

×

2220

  pRsp->tableType = TSDB_SUPER_TABLE;

×

2221

  pRsp->delay1 = pStb->maxdelay[0];

×

2222

  pRsp->delay2 = pStb->maxdelay[1];

×

2223

  pRsp->watermark1 = pStb->watermark[0];

×

2224

  pRsp->watermark2 = pStb->watermark[1];

×

2225

  pRsp->ttl = pStb->ttl;

×

2226

  pRsp->commentLen = pStb->commentLen;

×

2227

  if (pStb->commentLen > 0) {

×

2228

    pRsp->pComment = taosStrdup(pStb->comment);

×

2231

  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {

×

2232

    SSchema *pSchema = &pRsp->pSchemas[i];

×

2233

    SSchema *pSrcSchema = &pStb->pColumns[i];

×

2234

    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);

×

2235

    pSchema->type = pSrcSchema->type;

×

2236

    pSchema->flags = pSrcSchema->flags;

×

2237

    pSchema->colId = pSrcSchema->colId;

×

2238

    pSchema->bytes = pSrcSchema->bytes;

×

2241

  for (int32_t i = 0; i < pStb->numOfTags; ++i) {

×

2242

    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];

×

2243

    SSchema *pSrcSchema = &pStb->pTags[i];

×

2244

    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);

×

2245

    pSchema->type = pSrcSchema->type;

×

2246

    pSchema->flags = pSrcSchema->flags;

×

2247

    pSchema->colId = pSrcSchema->colId;

×

2248

    pSchema->bytes = pSrcSchema->bytes;

×

2251

  if (pStb->numOfFuncs > 0) {

×

2252

    pRsp->pFuncs = taosArrayDup(pStb->pFuncs, NULL);

×

2255

  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));

×

2256

  for (int32_t i = 0; i < pStb->numOfColumns; i++) {

×

2257

    SColCmpr *pCmpr = &pStb->pCmpr[i];

×

2259

    SSchemaExt *pSchExt = &pRsp->pSchemaExt[i];

×

2260

    pSchExt->colId = pCmpr->id;

×

2261

    pSchExt->compress = pCmpr->alg;

×

2264

  taosRUnLockLatch(&pStb->lock);

×

2265

  TAOS_RETURN(code);

×

2268

static int32_t mndValidateStbVersion(SMnode *pMnode, SSTableVersion *pStbVer, bool *schema, bool *sma) {

×

2269

  int32_t code = 0;

×

2270

  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2271

  snprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVer->dbFName, pStbVer->stbName);

×

2273

  SDbObj *pDb = mndAcquireDb(pMnode, pStbVer->dbFName);

×

2274

  if (pDb == NULL) {

×

2275

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2276

    TAOS_RETURN(code);

×

2279

  if (pDb->uid != pStbVer->dbId) {

×

2280

    mndReleaseDb(pMnode, pDb);

×

2281

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2282

    TAOS_RETURN(code);

×

2285

  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);

×

2286

  if (pStb == NULL) {

×

2287

    mndReleaseDb(pMnode, pDb);

×

2288

    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

2289

    TAOS_RETURN(code);

×

2292

  taosRLockLatch(&pStb->lock);

×

2294

  if (pStbVer->sversion != pStb->colVer || pStbVer->tversion != pStb->tagVer) {

×

2295

    *schema = true;

×

2297

    *schema = false;

×

2300

  if (pStbVer->smaVer && pStbVer->smaVer != pStb->smaVer) {

×

2301

    *sma = true;

×

2303

    *sma = false;

×

2306

  taosRUnLockLatch(&pStb->lock);

×

2308

  mndReleaseDb(pMnode, pDb);

×

2309

  mndReleaseStb(pMnode, pStb);

×

2310

  return TSDB_CODE_SUCCESS;

×

2313

static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {

×

2314

  int32_t code = 0;

×

2315

  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2316

  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);

×

2318

  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);

×

2319

  if (pDb == NULL) {

×

2320

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2321

    TAOS_RETURN(code);

×

2324

  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);

×

2325

  if (pStb == NULL) {

×

2326

    mndReleaseDb(pMnode, pDb);

×

2327

    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

2328

    TAOS_RETURN(code);

×

2331

  code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp);

×

2332

  mndReleaseDb(pMnode, pDb);

×

2333

  mndReleaseStb(pMnode, pStb);

×

2334

  TAOS_RETURN(code);

×

2337

static int32_t mndBuildStbCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {

×

2338

  int32_t code = 0;

×

2339

  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2340

  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);

×

2342

  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);

×

2343

  if (pDb == NULL) {

×

2344

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2345

    TAOS_RETURN(code);

×

2348

  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);

×

2349

  if (pStb == NULL) {

×

2350

    mndReleaseDb(pMnode, pDb);

×

2351

    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

2352

    TAOS_RETURN(code);

×

2355

  code = mndBuildStbCfgImp(pDb, pStb, tbName, pRsp);

×

2357

  mndReleaseDb(pMnode, pDb);

×

2358

  mndReleaseStb(pMnode, pStb);

×

2359

  TAOS_RETURN(code);

×

2362

static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, int32_t *pLen) {

×

2363

  int32_t       code = 0;

×

2364

  SEncoder      ec = {0};

×

2365

  uint32_t      contLen = 0;

×

2366

  SMAlterStbRsp alterRsp = {0};

×

2367

  SName         name = {0};

×

2368

  TAOS_CHECK_RETURN(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));

×

2370

  alterRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));

×

2371

  if (NULL == alterRsp.pMeta) {

×

2372

    code = terrno;

×

2373

    TAOS_RETURN(code);

×

2376

  code = mndBuildStbSchemaImp(pDb, pObj, name.tname, alterRsp.pMeta);

×

2377

  if (code) {

×

2378

    tFreeSMAlterStbRsp(&alterRsp);

×

2379

    return code;

×

2382

  tEncodeSize(tEncodeSMAlterStbRsp, &alterRsp, contLen, code);

×

2383

  if (code) {

×

2384

    tFreeSMAlterStbRsp(&alterRsp);

×

2385

    return code;

×

2388

  void *cont = taosMemoryMalloc(contLen);

×

2389

  if (NULL == cont) {

×

2390

    code = terrno;

×

2391

    tFreeSMAlterStbRsp(&alterRsp);

×

2392

    TAOS_RETURN(code);

×

2394

  tEncoderInit(&ec, cont, contLen);

×

2395

  code = tEncodeSMAlterStbRsp(&ec, &alterRsp);

×

2396

  tEncoderClear(&ec);

×

2398

  tFreeSMAlterStbRsp(&alterRsp);

×

2400

  if (code < 0) TAOS_RETURN(code);

×

2402

  *pCont = cont;

×

2403

  *pLen = contLen;

×

2405

  TAOS_RETURN(code);

×

2408

int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, void **pCont, int32_t *pLen) {

×

2409

  int32_t code = -1;

×

2410

  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);

×

2411

  if (NULL == pDb) {

×

2412

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2413

    if (terrno != 0) code = terrno;

×

2414

    TAOS_RETURN(code);

×

2417

  SStbObj *pObj = mndAcquireStb(pMnode, stbFName);

×

2418

  if (NULL == pObj) {

×

2419

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2420

    if (terrno != 0) code = terrno;

×

2421

    goto _OVER;

×

2424

  SEncoder       ec = {0};

×

2425

  uint32_t       contLen = 0;

×

2426

  SMCreateStbRsp stbRsp = {0};

×

2427

  SName          name = {0};

×

2428

  TAOS_CHECK_GOTO(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE), NULL, _OVER);

×

2430

  stbRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));

×

2431

  if (NULL == stbRsp.pMeta) {

×

2432

    code = terrno;

×

2433

    goto _OVER;

×

2436

  code = mndBuildStbSchemaImp(pDb, pObj, name.tname, stbRsp.pMeta);

×

2437

  if (code) {

×

2438

    tFreeSMCreateStbRsp(&stbRsp);

×

2439

    goto _OVER;

×

2442

  tEncodeSize(tEncodeSMCreateStbRsp, &stbRsp, contLen, code);

×

2443

  if (code) {

×

2444

    tFreeSMCreateStbRsp(&stbRsp);

×

2445

    goto _OVER;

×

2448

  void *cont = taosMemoryMalloc(contLen);

×

2449

  if (NULL == cont) {

×

2450

    code = terrno;

×

2451

    tFreeSMCreateStbRsp(&stbRsp);

×

2452

    goto _OVER;

×

2454

  tEncoderInit(&ec, cont, contLen);

×

2455

  TAOS_CHECK_GOTO(tEncodeSMCreateStbRsp(&ec, &stbRsp), NULL, _OVER);

×

2456

  tEncoderClear(&ec);

×

2458

  tFreeSMCreateStbRsp(&stbRsp);

×

2460

  *pCont = cont;

×

2461

  *pLen = contLen;

×

2463

  code = 0;

×

2465

_OVER:

×

2466

  if (pObj) {

×

2467

    mndReleaseStb(pMnode, pObj);

×

2470

  if (pDb) {

×

2471

    mndReleaseDb(pMnode, pDb);

×

2474

  TAOS_RETURN(code);

×

2477

static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,

×

2479

  int32_t code = -1;

×

2480

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");

×

2481

  if (pTrans == NULL) {

×

2482

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2483

    if (terrno != 0) code = terrno;

×

2484

    goto _OVER;

×

2487

  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);

×

2488

  mndTransSetDbName(pTrans, pDb->name, pStb->name);

×

2489

  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

×

2491

  if (needRsp) {

×

2492

    void   *pCont = NULL;

×

2493

    int32_t contLen = 0;

×

2494

    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);

×

2495

    mndTransSetRpcRsp(pTrans, pCont, contLen);

×

2498

  TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2499

  TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2500

  TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);

×

2501

  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);

×

2503

  code = 0;

×

2505

_OVER:

×

2506

  mndTransDrop(pTrans);

×

2507

  TAOS_RETURN(code);

×

2510

static int32_t mndAlterStbAndUpdateTagIdxImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,

×

2512

  int32_t code = -1;

×

2513

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");

×

2514

  if (pTrans == NULL) {

×

2515

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2516

    if (terrno != 0) code = terrno;

×

2517

    goto _OVER;

×

2520

  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);

×

2521

  mndTransSetDbName(pTrans, pDb->name, pStb->name);

×

2523

  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

×

2525

  if (needRsp) {

×

2526

    void   *pCont = NULL;

×

2527

    int32_t contLen = 0;

×

2528

    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);

×

2529

    mndTransSetRpcRsp(pTrans, pCont, contLen);

×

2532

  if (pAlter->alterType == TSDB_ALTER_TABLE_DROP_TAG) {

×

2533

    SIdxObj idxObj = {0};

×

2534

    SField *pField0 = taosArrayGet(pAlter->pFields, 0);

×

2535

    bool    exist = false;

×

2536

    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {

×

2537

      exist = true;

×

2539

    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2540

    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2542

    if (exist == true) {

×

2543

      TAOS_CHECK_GOTO(mndSetDropIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);

×

2544

      TAOS_CHECK_GOTO(mndSetDropIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);

×

2547

    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);

×

2548

    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);

×

2550

  } else if (pAlter->alterType == TSDB_ALTER_TABLE_UPDATE_TAG_NAME) {

×

2551

    SIdxObj     idxObj = {0};

×

2552

    SField     *pField0 = taosArrayGet(pAlter->pFields, 0);

×

2553

    SField     *pField1 = taosArrayGet(pAlter->pFields, 1);

×

2554

    const char *oTagName = pField0->name;

×

2555

    const char *nTagName = pField1->name;

×

2556

    bool        exist = false;

×

2558

    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {

×

2559

      exist = true;

×

2562

    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2563

    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2565

    if (exist == true) {

×

2566

      memcpy(idxObj.colName, nTagName, strlen(nTagName));

×

2567

      idxObj.colName[strlen(nTagName)] = 0;

×

2568

      TAOS_CHECK_GOTO(mndSetAlterIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);

×

2569

      TAOS_CHECK_GOTO(mndSetAlterIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);

×

2572

    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);

×

2573

    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);

×

2575

  code = 0;

×

2577

_OVER:

×

2578

  mndTransDrop(pTrans);

×

2579

  TAOS_RETURN(code);

×

2582

static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {

×

2583

  bool    needRsp = true;

×

2584

  int32_t code = -1;

×

2585

  SField *pField0 = NULL;

×

2587

  SStbObj stbObj = {0};

×

2588

  taosRLockLatch(&pOld->lock);

×

2589

  memcpy(&stbObj, pOld, sizeof(SStbObj));

×

2590

  taosRUnLockLatch(&pOld->lock);

×

2591

  stbObj.pColumns = NULL;

×

2592

  stbObj.pTags = NULL;

×

2593

  stbObj.pFuncs = NULL;

×

2594

  stbObj.pCmpr = NULL;

×

2595

  stbObj.updateTime = taosGetTimestampMs();

×

2596

  stbObj.lock = 0;

×

2597

  bool updateTagIndex = false;

×

2598

  switch (pAlter->alterType) {

×

2599

    case TSDB_ALTER_TABLE_ADD_TAG:

×

2600

      code = mndAddSuperTableTag(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);

×

2601

      break;

×

2602

    case TSDB_ALTER_TABLE_DROP_TAG:

×

2603

      pField0 = taosArrayGet(pAlter->pFields, 0);

×

2604

      code = mndDropSuperTableTag(pMnode, pOld, &stbObj, pField0->name);

×

2605

      updateTagIndex = true;

×

2606

      break;

×

2607

    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:

×

2608

      code = mndAlterStbTagName(pMnode, pOld, &stbObj, pAlter->pFields);

×

2609

      updateTagIndex = true;

×

2610

      break;

×

2611

    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:

×

2612

      pField0 = taosArrayGet(pAlter->pFields, 0);

×

2613

      code = mndAlterStbTagBytes(pMnode, pOld, &stbObj, pField0);

×

2614

      break;

×

2615

    case TSDB_ALTER_TABLE_ADD_COLUMN:

×

2616

      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields, 0);

×

2617

      break;

×

2618

    case TSDB_ALTER_TABLE_DROP_COLUMN:

×

2619

      pField0 = taosArrayGet(pAlter->pFields, 0);

×

2620

      code = mndDropSuperTableColumn(pMnode, pOld, &stbObj, pField0->name);

×

2621

      break;

×

2622

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:

×

2623

      pField0 = taosArrayGet(pAlter->pFields, 0);

×

2624

      code = mndAlterStbColumnBytes(pMnode, pOld, &stbObj, pField0);

×

2625

      break;

×

2626

    case TSDB_ALTER_TABLE_UPDATE_OPTIONS:

×

2627

      needRsp = false;

×

2628

      code = mndUpdateStbCommentAndTTL(pOld, &stbObj, pAlter->comment, pAlter->commentLen, pAlter->ttl);

×

2629

      break;

×

2630

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:

×

2631

      code = mndUpdateSuperTableColumnCompress(pMnode, pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);

×

2632

      break;

×

2633

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION:

×

2634

      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields, 1);

×

2635

      break;

×

2636

    default:

×

2637

      needRsp = false;

×

2638

      terrno = TSDB_CODE_OPS_NOT_SUPPORT;

×

2639

      break;

×

2642

  if (code != 0) goto _OVER;

×

2643

  if (updateTagIndex == false) {

×

2644

    code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen);

×

2646

    code = mndAlterStbAndUpdateTagIdxImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen, pAlter);

×

2649

_OVER:

×

2650

  taosMemoryFreeClear(stbObj.pTags);

×

2651

  taosMemoryFreeClear(stbObj.pColumns);

×

2652

  taosMemoryFreeClear(stbObj.pCmpr);

×

2653

  if (pAlter->commentLen > 0) {

×

2654

    taosMemoryFreeClear(stbObj.comment);

×

2656

  TAOS_RETURN(code);

×

2659

static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {

×

2660

  SMnode       *pMnode = pReq->info.node;

×

2661

  int32_t       code = -1;

×

2662

  SDbObj       *pDb = NULL;

×

2663

  SStbObj      *pStb = NULL;

×

2664

  SMAlterStbReq alterReq = {0};

×

2666

  if (tDeserializeSMAlterStbReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {

×

2667

    code = TSDB_CODE_INVALID_MSG;

×

2668

    goto _OVER;

×

2671

  mInfo("stb:%s, start to alter", alterReq.name);

×

2672

  if (mndCheckAlterStbReq(&alterReq) != 0) goto _OVER;

×

2674

  pDb = mndAcquireDbByStb(pMnode, alterReq.name);

×

2675

  if (pDb == NULL) {

×

2676

    code = TSDB_CODE_MND_DB_NOT_EXIST;

×

2677

    goto _OVER;

×

2680

  pStb = mndAcquireStb(pMnode, alterReq.name);

×

2681

  if (pStb == NULL) {

×

2682

    code = TSDB_CODE_MND_STB_NOT_EXIST;

×

2683

    goto _OVER;

×

2686

  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {

×

2687

    goto _OVER;

×

2690

  code = mndAlterStb(pMnode, pReq, &alterReq, pDb, pStb);

×

2691

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

2693

  SName   name = {0};

×

2694

  int32_t ret = 0;

×

2695

  if ((ret = tNameFromString(&name, alterReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)

×

2696

    mError("stb:%s, failed to tNameFromString since %s", alterReq.name, tstrerror(ret));

×

2698

  auditRecord(pReq, pMnode->clusterId, "alterStb", name.dbname, name.tname, alterReq.sql, alterReq.sqlLen);

×

2700

_OVER:

×

2701

  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

×

2702

    mError("stb:%s, failed to alter since %s", alterReq.name, tstrerror(code));

×

2705

  mndReleaseStb(pMnode, pStb);

×

2706

  mndReleaseDb(pMnode, pDb);

×

2707

  tFreeSMAltertbReq(&alterReq);

×

2709

  TAOS_RETURN(code);

×

2712

static int32_t mndSetDropStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {

×

2713

  int32_t  code = 0;

×

2714

  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);

×

2715

  if (pRedoRaw == NULL) {

×

2716

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2717

    if (terrno != 0) code = terrno;

×

2718

    TAOS_RETURN(code);

×

2720

  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {

×

2721

    sdbFreeRaw(pRedoRaw);

×

2722

    TAOS_RETURN(code);

×

2724

  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));

×

2726

  TAOS_RETURN(code);

×

2729

static int32_t mndSetDropStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {

×

2730

  int32_t  code = 0;

×

2731

  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);

×

2732

  if (pCommitRaw == NULL) {

×

2733

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2734

    if (terrno != 0) code = terrno;

×

2735

    TAOS_RETURN(code);

×

2737

  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {

×

2738

    sdbFreeRaw(pCommitRaw);

×

2739

    TAOS_RETURN(code);

×

2741

  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

×

2743

  TAOS_RETURN(code);

×

2746

static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {

×

2747

  int32_t code = 0;

×

2748

  SSdb   *pSdb = pMnode->pSdb;

×

2749

  SVgObj *pVgroup = NULL;

×

2750

  void   *pIter = NULL;

×

2752

  while (1) {

×

2753

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

2754

    if (pIter == NULL) break;

×

2755

    if (!mndVgroupInDb(pVgroup, pDb->uid)) {

×

2756

      sdbRelease(pSdb, pVgroup);

×

2757

      continue;

×

2760

    int32_t contLen = 0;

×

2761

    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);

×

2762

    if (pReq == NULL) {

×

2763

      sdbCancelFetch(pSdb, pIter);

×

2764

      sdbRelease(pSdb, pVgroup);

×

2765

      code = TSDB_CODE_OUT_OF_MEMORY;

×

2766

      TAOS_RETURN(code);

×

2769

    STransAction action = {0};

×

2770

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

2771

    action.pCont = pReq;

×

2772

    action.contLen = contLen;

×

2773

    action.msgType = TDMT_VND_DROP_STB;

×

2774

    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

2775

    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {

×

2776

      taosMemoryFree(pReq);

×

2777

      sdbCancelFetch(pSdb, pIter);

×

2778

      sdbRelease(pSdb, pVgroup);

×

2779

      TAOS_RETURN(code);

×

2781

    sdbRelease(pSdb, pVgroup);

×

2784

  TAOS_RETURN(code);

×

2787

static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb) {

×

2788

  int32_t code = -1;

×

2789

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "drop-stb");

×

2790

  if (pTrans == NULL) {

×

2791

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2792

    if (terrno != 0) code = terrno;

×

2793

    goto _OVER;

×

2796

  mInfo("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);

×

2797

  mndTransSetDbName(pTrans, pDb->name, pStb->name);

×

2798

  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

×

2800

  TAOS_CHECK_GOTO(mndSetDropStbPrepareLogs(pMnode, pTrans, pStb), NULL, _OVER);

×

2801

  TAOS_CHECK_GOTO(mndSetDropStbCommitLogs(pMnode, pTrans, pStb), NULL, _OVER);

×

2802

  TAOS_CHECK_GOTO(mndSetDropStbRedoActions(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2803

  TAOS_CHECK_GOTO(mndDropIdxsByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2804

  TAOS_CHECK_GOTO(mndDropSmasByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);

×

2805

  TAOS_CHECK_GOTO(mndUserRemoveStb(pMnode, pTrans, pStb->name), NULL, _OVER);

×

2806

  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);

×

2807

  code = 0;

×

2809

_OVER:

×

2810

  mndTransDrop(pTrans);

×

2811

  TAOS_RETURN(code);

×

2814

static int32_t mndCheckDropStbForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid) {

×

2815

  int32_t code = 0;

×

2816

  SSdb   *pSdb = pMnode->pSdb;

×

2817

  void   *pIter = NULL;

×

2818

  while (1) {

×

2819

    SMqTopicObj *pTopic = NULL;

×

2820

    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);

×

2821

    if (pIter == NULL) break;

×

2823

    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {

×

2824

      if (pTopic->stbUid == suid) {

×

2825

        sdbRelease(pSdb, pTopic);

×

2826

        sdbCancelFetch(pSdb, pIter);

×

2827

        TAOS_RETURN(-1);

×

2831

    if (pTopic->ast == NULL) {

×

2832

      sdbRelease(pSdb, pTopic);

×

2833

      continue;

×

2836

    SNode *pAst = NULL;

×

2837

    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {

×

2838

      code = TSDB_CODE_MND_INVALID_TOPIC_OPTION;

×

2839

      mError("topic:%s, create ast error", pTopic->name);

×

2840

      sdbRelease(pSdb, pTopic);

×

2841

      sdbCancelFetch(pSdb, pIter);

×

2842

      TAOS_RETURN(code);

×

2845

    SNodeList *pNodeList = NULL;

×

2846

    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=

×

2848

      sdbRelease(pSdb, pTopic);

×

2849

      sdbCancelFetch(pSdb, pIter);

×

2850

      TAOS_RETURN(code);

×

2852

    SNode *pNode = NULL;

×

2853

    FOREACH(pNode, pNodeList) {

×

2854

      SColumnNode *pCol = (SColumnNode *)pNode;

×

2856

      if (pCol->tableId == suid) {

×

2857

        sdbRelease(pSdb, pTopic);

×

2858

        nodesDestroyNode(pAst);

×

2859

        nodesDestroyList(pNodeList);

×

2860

        sdbCancelFetch(pSdb, pIter);

×

2861

        TAOS_RETURN(-1);

×

2863

        goto NEXT;

×

2866

  NEXT:

×

2867

    sdbRelease(pSdb, pTopic);

×

2868

    nodesDestroyNode(pAst);

×

2869

    nodesDestroyList(pNodeList);

×

2871

  TAOS_RETURN(code);

×

2874

static int32_t mndCheckDropStbForStream(SMnode *pMnode, const char *stbFullName, int64_t suid) {

×

2875

  int32_t code = 0;

×

2876

  SSdb   *pSdb = pMnode->pSdb;

×

2877

  void   *pIter = NULL;

×

2878

  while (1) {

×

2879

    SStreamObj *pStream = NULL;

×

2880

    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);

×

2881

    if (pIter == NULL) break;

×

2883

    if (pStream->targetStbUid == suid) {

×

2884

      sdbCancelFetch(pSdb, pIter);

×

2885

      sdbRelease(pSdb, pStream);

×

2886

      TAOS_RETURN(-1);

×

2889

    SNode *pAst = NULL;

×

2890

    if (nodesStringToNode(pStream->ast, &pAst) != 0) {

×

2891

      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;

×

2892

      mError("stream:%s, create ast error", pStream->name);

×

2893

      sdbCancelFetch(pSdb, pIter);

×

2894

      sdbRelease(pSdb, pStream);

×

2895

      TAOS_RETURN(code);

×

2898

    SNodeList *pNodeList = NULL;

×

2899

    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=

×

2901

      sdbCancelFetch(pSdb, pIter);

×

2902

      sdbRelease(pSdb, pStream);

×

2903

      TAOS_RETURN(code);

×

2905

    SNode *pNode = NULL;

×

2906

    FOREACH(pNode, pNodeList) {

×

2907

      SColumnNode *pCol = (SColumnNode *)pNode;

×

2909

      if (pCol->tableId == suid) {

×

2910

        sdbCancelFetch(pSdb, pIter);

×

2911

        sdbRelease(pSdb, pStream);

×

2912

        nodesDestroyNode(pAst);

×

2913

        nodesDestroyList(pNodeList);

×

2914

        TAOS_RETURN(-1);

×

2916

        goto NEXT;

×

2919

  NEXT:

×

2920

    sdbRelease(pSdb, pStream);

×

2921

    nodesDestroyNode(pAst);

×

2922

    nodesDestroyList(pNodeList);

×

2924

  TAOS_RETURN(code);

×

2927

static int32_t mndProcessDropTtltbRsp(SRpcMsg *pRsp) { return 0; }

×

2928

static int32_t mndProcessTrimDbRsp(SRpcMsg *pRsp) { return 0; }

×

2929

static int32_t mndProcessS3MigrateDbRsp(SRpcMsg *pRsp) { return 0; }

×

2931

static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {

×

2932

  SMnode      *pMnode = pReq->info.node;

×

2933

  int32_t      code = -1;

×

2934

  SDbObj      *pDb = NULL;

×

2935

  SStbObj     *pStb = NULL;

×

2936

  SMDropStbReq dropReq = {0};

×

2938

  TAOS_CHECK_GOTO(tDeserializeSMDropStbReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);

×

2940

  mInfo("stb:%s, start to drop", dropReq.name);

×

2942

  pStb = mndAcquireStb(pMnode, dropReq.name);

×

2943

  if (pStb == NULL) {

×

2944

    if (dropReq.igNotExists) {

×

2945

      mInfo("stb:%s, not exist, ignore not exist is set", dropReq.name);

×

2946

      code = 0;

×

2947

      goto _OVER;

×

2949

      code = TSDB_CODE_MND_STB_NOT_EXIST;

×

2950

      goto _OVER;

×

2954

  if ((dropReq.source == TD_REQ_FROM_TAOX_OLD || dropReq.source == TD_REQ_FROM_TAOX) && pStb->uid != dropReq.suid) {

×

2955

    code = 0;

×

2956

    goto _OVER;

×

2959

  pDb = mndAcquireDbByStb(pMnode, dropReq.name);

×

2960

  if (pDb == NULL) {

×

2961

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2962

    goto _OVER;

×

2965

  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {

×

2966

    goto _OVER;

×

2969

  if (mndCheckDropStbForTopic(pMnode, dropReq.name, pStb->uid) < 0) {

×

2970

    code = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;

×

2971

    goto _OVER;

×

2974

  if (mndCheckDropStbForStream(pMnode, dropReq.name, pStb->uid) < 0) {

×

2975

    code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;

×

2976

    goto _OVER;

×

2979

  code = mndDropStb(pMnode, pReq, pDb, pStb);

×

2980

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

2982

  SName   name = {0};

×

2983

  int32_t ret = 0;

×

2984

  if ((ret = tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)

×

2985

    mError("stb:%s, failed to tNameFromString since %s", dropReq.name, tstrerror(ret));

×

2987

  auditRecord(pReq, pMnode->clusterId, "dropStb", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen);

×

2989

_OVER:

×

2990

  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

×

2991

    mError("stb:%s, failed to drop since %s", dropReq.name, tstrerror(code));

×

2994

  mndReleaseDb(pMnode, pDb);

×

2995

  mndReleaseStb(pMnode, pStb);

×

2996

  tFreeSMDropStbReq(&dropReq);

×

2997

  TAOS_RETURN(code);

×

3000

static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {

×

3001

  SMnode       *pMnode = pReq->info.node;

×

3002

  int32_t       code = -1;

×

3003

  STableInfoReq infoReq = {0};

×

3004

  STableMetaRsp metaRsp = {0};

×

3005

  SUserObj     *pUser = NULL;

×

3007

  code = mndAcquireUser(pMnode, pReq->info.conn.user, &pUser);

×

3008

  if (pUser == NULL) return 0;

×

3009

  bool sysinfo = pUser->sysInfo;

×

3011

  TAOS_CHECK_GOTO(tDeserializeSTableInfoReq(pReq->pCont, pReq->contLen, &infoReq), NULL, _OVER);

×

3013

  if (0 == strcmp(infoReq.dbFName, TSDB_INFORMATION_SCHEMA_DB)) {

×

3014

    mInfo("information_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);

×

3015

    TAOS_CHECK_GOTO(mndBuildInsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, sysinfo, &metaRsp), NULL, _OVER);

×

3016

  } else if (0 == strcmp(infoReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {

×

3017

    mInfo("performance_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);

×

3018

    TAOS_CHECK_GOTO(mndBuildPerfsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);

×

3020

    mInfo("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);

×

3021

    TAOS_CHECK_GOTO(mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);

×

3024

  int32_t rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);

×

3025

  if (rspLen < 0) {

×

3026

    code = TSDB_CODE_INVALID_MSG;

×

3027

    goto _OVER;

×

3030

  void *pRsp = rpcMallocCont(rspLen);

×

3031

  if (pRsp == NULL) {

×

3032

    code = terrno;

×

3033

    goto _OVER;

×

3036

  if ((rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp)) < 0) {

×

3037

    code = rspLen;

×

3038

    goto _OVER;

×

3040

  pReq->info.rsp = pRsp;

×

3041

  pReq->info.rspLen = rspLen;

×

3042

  code = 0;

×

3044

  mTrace("%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);

×

3046

_OVER:

×

3047

  if (code != 0) {

×

3048

    mError("stb:%s.%s, failed to retrieve meta since %s", infoReq.dbFName, infoReq.tbName, tstrerror(code));

×

3051

  mndReleaseUser(pMnode, pUser);

×

3052

  tFreeSTableMetaRsp(&metaRsp);

×

3054

  return code;

×

3057

static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {

×

3058

  SMnode      *pMnode = pReq->info.node;

×

3059

  int32_t      code = -1;

×

3060

  STableCfgReq cfgReq = {0};

×

3061

  STableCfgRsp cfgRsp = {0};

×

3063

  TAOS_CHECK_GOTO(tDeserializeSTableCfgReq(pReq->pCont, pReq->contLen, &cfgReq), NULL, _OVER);

×

3065

  char dbName[TSDB_DB_NAME_LEN] = {0};

×

3066

  TAOS_CHECK_GOTO(mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName), NULL, _OVER);

×

3067

  if (0 == strcmp(dbName, TSDB_INFORMATION_SCHEMA_DB)) {

×

3068

    mInfo("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);

×

3069

    TAOS_CHECK_GOTO(mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);

×

3070

  } else if (0 == strcmp(dbName, TSDB_PERFORMANCE_SCHEMA_DB)) {

×

3071

    mInfo("performance_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);

×

3072

    TAOS_CHECK_GOTO(mndBuildPerfsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);

×

3074

    mInfo("stb:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);

×

3075

    TAOS_CHECK_GOTO(mndBuildStbCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);

×

3078

  int32_t rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);

×

3079

  if (rspLen < 0) {

×

3080

    code = TSDB_CODE_INVALID_MSG;

×

3081

    goto _OVER;

×

3084

  void *pRsp = rpcMallocCont(rspLen);

×

3085

  if (pRsp == NULL) {

×

3086

    code = terrno;

×

3087

    goto _OVER;

×

3090

  if ((rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp)) < 0) {

×

3091

    code = rspLen;

×

3092

    goto _OVER;

×

3094

  pReq->info.rsp = pRsp;

×

3095

  pReq->info.rspLen = rspLen;

×

3096

  code = 0;

×

3098

  mTrace("%s.%s, cfg is retrieved", cfgReq.dbFName, cfgReq.tbName);

×

3100

_OVER:

×

3101

  if (code != 0) {

×

3102

    mError("stb:%s.%s, failed to retrieve cfg since %s", cfgReq.dbFName, cfgReq.tbName, tstrerror(code));

×

3105

  tFreeSTableCfgRsp(&cfgRsp);

×

3106

  TAOS_RETURN(code);

×

3109

int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,

×

3111

  int32_t   code = 0;

×

3112

  SSTbHbRsp hbRsp = {0};

×

3113

  hbRsp.pMetaRsp = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));

×

3114

  if (hbRsp.pMetaRsp == NULL) {

×

3115

    code = terrno;

×

3116

    TAOS_RETURN(code);

×

3119

  hbRsp.pIndexRsp = taosArrayInit(numOfStbs, sizeof(STableIndexRsp));

×

3120

  if (NULL == hbRsp.pIndexRsp) {

×

3121

    taosArrayDestroy(hbRsp.pMetaRsp);

×

3122

    code = terrno;

×

3123

    TAOS_RETURN(code);

×

3126

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

×

3127

    SSTableVersion *pStbVersion = &pStbVersions[i];

×

3128

    pStbVersion->suid = be64toh(pStbVersion->suid);

×

3129

    pStbVersion->sversion = ntohl(pStbVersion->sversion);

×

3130

    pStbVersion->tversion = ntohl(pStbVersion->tversion);

×

3131

    pStbVersion->smaVer = ntohl(pStbVersion->smaVer);

×

3133

    bool    schema = false;

×

3134

    bool    sma = false;

×

3135

    int32_t code = mndValidateStbVersion(pMnode, pStbVersion, &schema, &sma);

×

3136

    if (TSDB_CODE_SUCCESS != code) {

×

3137

      STableMetaRsp metaRsp = {0};

×

3138

      metaRsp.numOfColumns = -1;

×

3139

      metaRsp.suid = pStbVersion->suid;

×

3140

      tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));

×

3141

      tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));

×

3142

      tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));

×

3143

      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {

×

3144

        code = terrno;

×

3145

        return code;

×

3147

      continue;

×

3150

    if (schema) {

×

3151

      STableMetaRsp metaRsp = {0};

×

3152

      mInfo("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);

×

3153

      if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp) != 0) {

×

3154

        metaRsp.numOfColumns = -1;

×

3155

        metaRsp.suid = pStbVersion->suid;

×

3156

        tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));

×

3157

        tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));

×

3158

        tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));

×

3159

        if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {

×

3160

          code = terrno;

×

3161

          return code;

×

3163

        continue;

×

3166

      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {

×

3167

        code = terrno;

×

3168

        return code;

×

3172

    if (sma) {

×

3173

      bool           exist = false;

×

3175

      STableIndexRsp indexRsp = {0};

×

3176

      indexRsp.pIndex = taosArrayInit(10, sizeof(STableIndexInfo));

×

3177

      if (NULL == indexRsp.pIndex) {

×

3178

        code = terrno;

×

3179

        TAOS_RETURN(code);

×

3182

      (void)tsnprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVersion->dbFName, pStbVersion->stbName);

×

3183

      int32_t code = mndGetTableSma(pMnode, tbFName, &indexRsp, &exist);

×

3184

      if (code || !exist) {

×

3185

        indexRsp.suid = pStbVersion->suid;

×

3186

        indexRsp.version = -1;

×

3187

        indexRsp.pIndex = NULL;

×

3190

      tstrncpy(indexRsp.dbFName, pStbVersion->dbFName, sizeof(indexRsp.dbFName));

×

3191

      tstrncpy(indexRsp.tbName, pStbVersion->stbName, sizeof(indexRsp.tbName));

×

3193

      if (taosArrayPush(hbRsp.pIndexRsp, &indexRsp) == NULL) {

×

3194

        code = terrno;

×

3195

        return code;

×

3200

  int32_t rspLen = tSerializeSSTbHbRsp(NULL, 0, &hbRsp);

×

3201

  if (rspLen < 0) {

×

3202

    tFreeSSTbHbRsp(&hbRsp);

×

3203

    code = TSDB_CODE_INVALID_MSG;

×

3204

    TAOS_RETURN(code);

×

3207

  void *pRsp = taosMemoryMalloc(rspLen);

×

3208

  if (pRsp == NULL) {

×

3209

    tFreeSSTbHbRsp(&hbRsp);

×

3210

    code = terrno;

×

3211

    TAOS_RETURN(code);

×

3214

  rspLen = tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);

×

3215

  tFreeSSTbHbRsp(&hbRsp);

×

3216

  if (rspLen < 0) return rspLen;

×

3217

  *ppRsp = pRsp;

×

3218

  *pRspLen = rspLen;

×

3219

  TAOS_RETURN(code);

×

3222

int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs) {

×

3223

  int32_t code = 0;

×

3224

  SSdb   *pSdb = pMnode->pSdb;

×

3225

  SDbObj *pDb = mndAcquireDb(pMnode, dbName);

×

3226

  if (pDb == NULL) {

×

3227

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

3228

    TAOS_RETURN(code);

×

3231

  int32_t numOfStbs = 0;

×

3232

  void   *pIter = NULL;

×

3233

  while (1) {

×

3234

    SStbObj *pStb = NULL;

×

3235

    pIter = sdbFetch(pSdb, SDB_STB, pIter, (void **)&pStb);

×

3236

    if (pIter == NULL) break;

×

3238

    if (pStb->dbUid == pDb->uid) {

×

3239

      numOfStbs++;

×

3242

    sdbRelease(pSdb, pStb);

×

3245

  *pNumOfStbs = numOfStbs;

×

3246

  mndReleaseDb(pMnode, pDb);

×

3247

  TAOS_RETURN(code);

×

3250

int32_t mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst) {

×

3251

  SName name = {0};

×

3252

  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));

×

3254

  TAOS_CHECK_RETURN(tNameGetFullDbName(&name, dst));

×

3256

  return 0;

×

3259

int32_t mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {

×

3260

  SName name = {0};

×

3261

  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));

×

3263

  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));

×

3265

  return 0;

×

3268

int32_t mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {

×

3269

  SName name = {0};

×

3270

  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB));

×

3272

  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));

×

3274

  return 0;

×

3277

void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize) {

×

3278

  int32_t pos = -1;

×

3279

  int32_t num = 0;

×

3280

  for (pos = 0; stbFullName[pos] != 0; ++pos) {

×

3281

    if (stbFullName[pos] == TS_PATH_DELIMITER[0]) num++;

×

3282

    if (num == 2) break;

×

3285

  if (num == 2) {

×

3286

    tstrncpy(dst, stbFullName + pos + 1, dstSize);

×

3288

×

3290

static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {

×

3291

  SMnode  *pMnode = pReq->info.node;

×

3292

  SSdb    *pSdb = pMnode->pSdb;

×

3293

  int32_t  numOfRows = 0;

×

3294

  SStbObj *pStb = NULL;

×

3295

  int32_t  cols = 0;

×

3296

  int32_t  lino = 0;

×

3297

  int32_t  code = 0;

×

3299

  SDbObj *pDb = NULL;

×

3300

  if (strlen(pShow->db) > 0) {

×

3301

    pDb = mndAcquireDb(pMnode, pShow->db);

×

3302

    if (pDb == NULL) return terrno;

×

3305

  while (numOfRows < rows) {

×

3306

    pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);

×

3307

    if (pShow->pIter == NULL) break;

×

3309

    if (pDb != NULL && pStb->dbUid != pDb->uid) {

×

3310

      sdbRelease(pSdb, pStb);

×

3311

      continue;

×

3314

    if (isTsmaResSTb(pStb->name)) {

×

3315

      sdbRelease(pSdb, pStb);

×

3316

      continue;

×

3319

    cols = 0;

×

3321

    SName name = {0};

×

3323

    char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3324

    mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);

×

3325

    varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));

×

3326

    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3327

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _ERROR);

×

3329

    char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3330

    RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _ERROR);

×

3331

    RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _ERROR);

×

3332

    varDataSetLen(db, strlen(varDataVal(db)));

×

3333

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3334

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _ERROR);

×

3336

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3337

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->createdTime, false), pStb, &lino,

×

3340

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3341

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false), pStb, &lino,

×

3344

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3345

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false), pStb, &lino, _ERROR);

×

3347

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3348

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->updateTime, false), pStb, &lino,

×

3351

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3352

    if (pStb->commentLen > 0) {

×

3353

      char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};

×

3354

      STR_TO_VARSTR(comment, pStb->comment);

×

3355

      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);

×

3356

    } else if (pStb->commentLen == 0) {

×

3357

      char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};

×

3358

      STR_TO_VARSTR(comment, "");

×

3359

      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);

×

3361

      colDataSetNULL(pColInfo, numOfRows);

×

3364

    char watermark[64 + VARSTR_HEADER_SIZE] = {0};

×

3365

    (void)tsnprintf(varDataVal(watermark), sizeof(watermark) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",

×

3366

              pStb->watermark[0], pStb->watermark[1]);

×

3367

    varDataSetLen(watermark, strlen(varDataVal(watermark)));

×

3369

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3370

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false), pStb, &lino, _ERROR);

×

3372

    char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};

×

3373

    (void)tsnprintf(varDataVal(maxDelay), sizeof(maxDelay) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",

×

3374

              pStb->maxdelay[0], pStb->maxdelay[1]);

×

3375

    varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));

×

3377

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3378

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false), pStb, &lino, _ERROR);

×

3380

    char    rollup[160 + VARSTR_HEADER_SIZE] = {0};

×

3381

    int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);

×

3382

    char   *sep = ", ";

×

3383

    int32_t sepLen = strlen(sep);

×

3384

    int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;

×

3385

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

×

3386

      char *funcName = taosArrayGet(pStb->pFuncs, i);

×

3387

      if (i) {

×

3388

        (void)strncat(varDataVal(rollup), sep, rollupLen);

×

3389

        rollupLen -= sepLen;

×

3391

      (void)strncat(varDataVal(rollup), funcName, rollupLen);

×

3392

      rollupLen -= strlen(funcName);

×

3394

    varDataSetLen(rollup, strlen(varDataVal(rollup)));

×

3396

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3397

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);

×

3399

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3400

    if (pColInfo) {

×

3401

      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->uid), false), pStb, &lino, _ERROR);

×

3404

    numOfRows++;

×

3405

    sdbRelease(pSdb, pStb);

×

3408

  if (pDb != NULL) {

×

3409

    mndReleaseDb(pMnode, pDb);

×

3412

  goto _OVER;

×

3414

_ERROR:

×

3415

  mError("show:0x%" PRIx64 ", failed to retrieve data at %s:%d since %s", pShow->id, __FUNCTION__, lino,

×

3418

_OVER:

×

3419

  pShow->numOfRows += numOfRows;

×

3420

  return numOfRows;

×

3423

static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *pSysDbTableMeta, size_t size,

×

3425

  char    tName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3426

  char    dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3427

  char    typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3428

  int32_t numOfRows = p->info.rows;

×

3429

  int32_t lino = 0;

×

3430

  int32_t code = 0;

×

3432

  STR_TO_VARSTR(dName, dbName);

×

3433

  STR_TO_VARSTR(typeName, "SYSTEM_TABLE");

×

3435

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

×

3436

    const SSysTableMeta *pm = &pSysDbTableMeta[i];

×

3440

    if (tbName[0] && strncmp(tbName, pm->name, TSDB_TABLE_NAME_LEN) != 0) {

×

3441

      continue;

×

3444

    STR_TO_VARSTR(tName, pm->name);

×

3446

    for (int32_t j = 0; j < pm->colNum; j++) {

×

3448

      SColumnInfoData *pColInfoData = taosArrayGet(p->pDataBlock, 0);

×

3449

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, tName, false), &lino, _OVER);

×

3452

      pColInfoData = taosArrayGet(p->pDataBlock, 1);

×

3453

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, dName, false), &lino, _OVER);

×

3455

      pColInfoData = taosArrayGet(p->pDataBlock, 2);

×

3456

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, typeName, false), &lino, _OVER);

×

3459

      char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3460

      STR_TO_VARSTR(colName, pm->schema[j].name);

×

3461

      pColInfoData = taosArrayGet(p->pDataBlock, 3);

×

3462

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, colName, false), &lino, _OVER);

×

3465

      int8_t colType = pm->schema[j].type;

×

3466

      pColInfoData = taosArrayGet(p->pDataBlock, 4);

×

3468

      int  colTypeLen =

×

3469

          tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);

×

3470

      if (colType == TSDB_DATA_TYPE_VARCHAR) {

×

3471

        colTypeLen +=

×

3472

            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",

×

3473

                      (int32_t)(pm->schema[j].bytes - VARSTR_HEADER_SIZE));

×

3474

      } else if (colType == TSDB_DATA_TYPE_NCHAR) {

×

3475

        colTypeLen +=

×

3476

            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",

×

3477

                      (int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));

×

3479

      varDataSetLen(colTypeStr, colTypeLen);

×

3480

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false), &lino, _OVER);

×

3482

      pColInfoData = taosArrayGet(p->pDataBlock, 5);

×

3483

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false), &lino, _OVER);

×

3484

      for (int32_t k = 6; k <= 8; ++k) {

×

3485

        pColInfoData = taosArrayGet(p->pDataBlock, k);

×

3486

        colDataSetNULL(pColInfoData, numOfRows);

×

3489

      numOfRows += 1;

×

3492

_OVER:

×

3493

  mError("failed at %s:%d since %s", __FUNCTION__, lino, tstrerror(code));

×

3494

  return numOfRows;

×

3501

static int32_t buildSysDbColsInfo(SSDataBlock *p, int8_t buildWhichDBs, char *tb) {

×

3502

  size_t               size = 0;

×

3503

  const SSysTableMeta *pSysDbTableMeta = NULL;

×

3505

  if (buildWhichDBs & BUILD_COL_FOR_INFO_DB) {

×

3506

    getInfosDbMeta(&pSysDbTableMeta, &size);

×

3507

    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB, tb);

×

3510

  if (buildWhichDBs & BUILD_COL_FOR_PERF_DB) {

×

3511

    getPerfDbMeta(&pSysDbTableMeta, &size);

×

3512

    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB, tb);

×

3515

  return p->info.rows;

×

3518

static int8_t determineBuildColForWhichDBs(const char *db) {

×

3520

  if (!db[0])

×

3521

    buildWhichDBs = BUILD_COL_FOR_ALL_DB;

×

3523

    char *p = strchr(db, '.');

×

3524

    if (p && strcmp(p + 1, TSDB_INFORMATION_SCHEMA_DB) == 0) {

×

3525

      buildWhichDBs = BUILD_COL_FOR_INFO_DB;

×

3526

    } else if (p && strcmp(p + 1, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

×

3527

      buildWhichDBs = BUILD_COL_FOR_PERF_DB;

×

3529

      buildWhichDBs = BUILD_COL_FOR_USER_DB;

×

3532

  return buildWhichDBs;

×

3535

static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {

×

3537

  SMnode  *pMnode = pReq->info.node;

×

3538

  SSdb    *pSdb = pMnode->pSdb;

×

3539

  SStbObj *pStb = NULL;

×

3540

  int32_t  numOfRows = 0;

×

3541

  int32_t  lino = 0;

×

3542

  int32_t  code = 0;

×

3544

  buildWhichDBs = determineBuildColForWhichDBs(pShow->db);

×

3546

  if (!pShow->sysDbRsp) {

×

3547

    numOfRows = buildSysDbColsInfo(pBlock, buildWhichDBs, pShow->filterTb);

×

3548

    mDebug("mndRetrieveStbCol get system table cols, rows:%d, db:%s", numOfRows, pShow->db);

×

3549

    pShow->sysDbRsp = true;

×

3552

  if (buildWhichDBs & BUILD_COL_FOR_USER_DB) {

×

3553

    SDbObj *pDb = NULL;

×

3554

    if (strlen(pShow->db) > 0) {

×

3555

      pDb = mndAcquireDb(pMnode, pShow->db);

×

3556

      if (pDb == NULL && TSDB_CODE_MND_DB_NOT_EXIST != terrno && pBlock->info.rows == 0) return terrno;

×

3559

    char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3560

    STR_TO_VARSTR(typeName, "SUPER_TABLE");

×

3561

    bool fetch = pShow->restore ? false : true;

×

3562

    pShow->restore = false;

×

3563

    while (numOfRows < rows) {

×

3564

      if (fetch) {

×

3565

        pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);

×

3566

        if (pShow->pIter == NULL) break;

×

3568

        fetch = true;

×

3569

        void *pKey = taosHashGetKey(pShow->pIter, NULL);

×

3570

        pStb = sdbAcquire(pSdb, SDB_STB, pKey);

×

3571

        if (!pStb) continue;

×

3574

      if (pDb != NULL && pStb->dbUid != pDb->uid) {

×

3575

        sdbRelease(pSdb, pStb);

×

3576

        continue;

×

3579

      SName name = {0};

×

3580

      char  stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3581

      mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);

×

3582

      if (pShow->filterTb[0] && strncmp(pShow->filterTb, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN) != 0) {

×

3583

        sdbRelease(pSdb, pStb);

×

3584

        continue;

×

3587

      if ((numOfRows + pStb->numOfColumns) > rows) {

×

3588

        pShow->restore = true;

×

3589

        if (numOfRows == 0) {

×

3590

          mError("mndRetrieveStbCol failed to get stable cols since buf:%d less than result:%d, stable name:%s, db:%s",

×

3593

        sdbRelease(pSdb, pStb);

×

3594

        break;

×

3597

      varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));

×

3599

      mDebug("mndRetrieveStbCol get stable cols, stable name:%s, db:%s", pStb->name, pStb->db);

×

3601

      char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3602

      RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _OVER);

×

3603

      RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _OVER);

×

3604

      varDataSetLen(db, strlen(varDataVal(db)));

×

3606

      for (int i = 0; i < pStb->numOfColumns; i++) {

×

3607

        int32_t          cols = 0;

×

3608

        SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3609

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _OVER);

×

3611

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3612

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _OVER);

×

3614

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3615

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeName, false), pStb, &lino, _OVER);

×

3618

        char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3619

        STR_TO_VARSTR(colName, pStb->pColumns[i].name);

×

3620

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3621

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, colName, false), pStb, &lino, _OVER);

×

3624

        int8_t colType = pStb->pColumns[i].type;

×

3625

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3627

        int  colTypeLen =

×

3628

            tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);

×

3629

        if (colType == TSDB_DATA_TYPE_VARCHAR) {

×

3630

          colTypeLen +=

×

3631

              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,

×

3632

                        "(%d)", (int32_t)(pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE));

×

3633

        } else if (colType == TSDB_DATA_TYPE_NCHAR) {

×

3634

          colTypeLen +=

×

3635

              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,

×

3636

                        "(%d)", (int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));

×

3638

        varDataSetLen(colTypeStr, colTypeLen);

×

3639

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false), pStb, &lino, _OVER);

×

3641

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3642

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false), pStb,

×

3644

        while (cols < pShow->numOfColumns) {

×

3645

          pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3646

          colDataSetNULL(pColInfo, numOfRows);

×

3648

        numOfRows++;

×

3651

      sdbRelease(pSdb, pStb);

×

3654

    if (pDb != NULL) {

×

3655

      mndReleaseDb(pMnode, pDb);

×

3659

  mDebug("mndRetrieveStbCol success, rows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows);

×

3660

  goto _OVER;

×

3666

_OVER:

×

3667

  pShow->numOfRows += numOfRows;

×

3668

  return numOfRows;

×

3671

static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {

×

3672

  SSdb *pSdb = pMnode->pSdb;

×

3673

  sdbCancelFetchByType(pSdb, pIter, SDB_STB);

×

3674

×

3676

const char *mndGetStbStr(const char *src) {

×

3677

  char *posDb = strstr(src, TS_PATH_DELIMITER);

×

3678

  if (posDb != NULL) ++posDb;

×

3679

  if (posDb == NULL) return src;

×

3681

  char *posStb = strstr(posDb, TS_PATH_DELIMITER);

×

3682

  if (posStb != NULL) ++posStb;

×

3683

  if (posStb == NULL) return posDb;

×

3684

  return posStb;

×

3687

static int32_t mndCheckIndexReq(SCreateTagIndexReq *pReq) {

×

3689

  return TSDB_CODE_SUCCESS;

×

3816

static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq) {

×

3817

  int32_t code = mndProcessDropStbReq(pReq);

×

3818

  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

×

3819

    pReq->info.rsp = rpcMallocCont(1);

×

3820

    pReq->info.rspLen = 1;

×

3821

    pReq->info.noResp = false;

×

3822

    pReq->code = code;

×

3824

  return code;

×

3857

static void mndDestroyDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx *p) {

×

3858

  if (!p) return;

×

3860

  if (p->pVgMap) {

×

3861

    void *pIter = taosHashIterate(p->pVgMap, NULL);

×

3862

    while (pIter) {

×

3863

      SVDropTbVgReqs *pReqs = pIter;

×

3864

      taosArrayDestroyEx(pReqs->pBatchReqs, destroySVDropTbBatchReqs);

×

3865

      pIter = taosHashIterate(p->pVgMap, pIter);

×

3867

    taosHashCleanup(p->pVgMap);

×

3869

  taosMemoryFree(p);

×

3872

static int32_t mndInitDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx **ppCtx) {

×

3873

  int32_t                 code = 0;

×

3874

  SMndDropTbsWithTsmaCtx *pCtx = taosMemoryCalloc(1, sizeof(SMndDropTbsWithTsmaCtx));

×

3875

  if (!pCtx) return terrno;

×

3877

  pCtx->pVgMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);

×

3878

  if (!pCtx->pVgMap) {

×

3879

    code = terrno;

×

3880

    goto _end;

×

3883

  *ppCtx = pCtx;

×

3884

_end:

×

3885

  if (code) mndDestroyDropTbsWithTsmaCtx(pCtx);

×

3886

  return code;

×

3889

static void *mndBuildVDropTbsReq(SMnode *pMnode, const SVgroupInfo *pVgInfo, const SVDropTbBatchReq *pReq,

×

3891

  int32_t   contLen = 0;

×

3892

  int32_t   ret = 0;

×

3893

  SMsgHead *pHead = NULL;

×

3894

  SEncoder  encoder = {0};

×

3896

  tEncodeSize(tEncodeSVDropTbBatchReq, pReq, contLen, ret);

×

3897

  if (ret < 0) return NULL;

×

3899

  contLen += sizeof(SMsgHead);

×

3900

  pHead = taosMemoryMalloc(contLen);

×

3901

  if (pHead == NULL) {

×

3902

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

3903

    return NULL;

×

3906

  pHead->contLen = htonl(contLen);

×

3907

  pHead->vgId = htonl(pVgInfo->vgId);

×

3909

  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));

×

3911

  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));

×

3912

  int32_t code = tEncodeSVDropTbBatchReq(&encoder, pReq);

×

3913

  tEncoderClear(&encoder);

×

3914

  if (code != 0) return NULL;

×

3916

  *len = contLen;

×

3917

  return pHead;

×

3920

static int32_t mndSetDropTbsRedoActions(SMnode *pMnode, STrans *pTrans, const SVDropTbVgReqs *pVgReqs, void *pCont,

×

3922

  STransAction action = {0};

×

3923

  action.epSet = pVgReqs->info.epSet;

×

3924

  action.pCont = pCont;

×

3925

  action.contLen = contLen;

×

3926

  action.msgType = msgType;

×

3927

  action.acceptableCode = TSDB_CODE_TDB_TABLE_NOT_EXIST;

×

3928

  return mndTransAppendRedoAction(pTrans, &action);

×

3931

static int32_t mndBuildDropTbRedoActions(SMnode *pMnode, STrans *pTrans, SHashObj *pVgMap, tmsg_t msgType) {

×

3932

  int32_t code = 0;

×

3933

  void   *pIter = taosHashIterate(pVgMap, NULL);

×

3934

  while (pIter) {

×

3935

    const SVDropTbVgReqs *pVgReqs = pIter;

×

3936

    int32_t               len = 0;

×

3937

    for (int32_t i = 0; i < taosArrayGetSize(pVgReqs->pBatchReqs) && code == TSDB_CODE_SUCCESS; ++i) {

×

3938

      SVDropTbBatchReq *pBatchReq = taosArrayGet(pVgReqs->pBatchReqs, i);

×

3939

      void             *p = mndBuildVDropTbsReq(pMnode, &pVgReqs->info, pBatchReq, &len);

×

3940

      if (!p) {

×

3941

        code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

3942

        if (terrno != 0) code = terrno;

×

3943

        break;

×

3945

      if ((code = mndSetDropTbsRedoActions(pMnode, pTrans, pVgReqs, p, len, msgType)) != 0) {

×

3946

        break;

×

3949

    if (TSDB_CODE_SUCCESS != code) {

×

3950

      taosHashCancelIterate(pVgMap, pIter);

×

3951

      break;

×

3953

    pIter = taosHashIterate(pVgMap, pIter);

×

3955

  return code;

×

3958

static int32_t mndCreateDropTbsTxnPrepare(SRpcMsg *pRsp, SMndDropTbsWithTsmaCtx *pCtx) {

×

3959

  int32_t code = 0;

×

3960

  SMnode *pMnode = pRsp->info.node;

×

3961

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pRsp, "drop-tbs");

×

3962

  mndTransSetChangeless(pTrans);

×

3963

  mndTransSetSerial(pTrans);

×

3964

  if (pTrans == NULL) {

×

3965

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

3966

    if (terrno != 0) code = terrno;

×

3967

    goto _OVER;

×

3970

  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

×

3972

  if ((code = mndBuildDropTbRedoActions(pMnode, pTrans, pCtx->pVgMap, TDMT_VND_DROP_TABLE)) != 0) goto _OVER;

×

3973

  if ((code = mndTransPrepare(pMnode, pTrans)) != 0) goto _OVER;

×

3975

_OVER:

×

3976

  mndTransDrop(pTrans);

×

3977

  TAOS_RETURN(code);

×

3980

static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq) {

×

3981

  int32_t      code = -1;

×

3982

  SMnode      *pMnode = pReq->info.node;

×

3983

  SDbObj      *pDb = NULL;

×

3984

  SStbObj     *pStb = NULL;

×

3985

  SMDropTbsReq dropReq = {0};

×

3986

  bool         locked = false;

×

3987

  if (tDeserializeSMDropTbsReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {

×

3988

    code = TSDB_CODE_INVALID_MSG;

×

3989

    goto _OVER;

×

3992

  SMndDropTbsWithTsmaCtx *pCtx = NULL;

×

3993

  code = mndInitDropTbsWithTsmaCtx(&pCtx);

×

3994

  if (code) goto _OVER;

×

3995

  for (int32_t i = 0; i < dropReq.pVgReqs->size; ++i) {

×

3996

    SMDropTbReqsOnSingleVg *pReq = taosArrayGet(dropReq.pVgReqs, i);

×

3997

    code = mndDropTbForSingleVg(pMnode, pCtx, pReq->pTbs, pReq->vgInfo.vgId);

×

3998

    if (code) goto _OVER;

×

4000

  code = mndCreateDropTbsTxnPrepare(pReq, pCtx);

×

4001

  if (code == 0) {

×

4002

    code = TSDB_CODE_ACTION_IN_PROGRESS;

×

4004

_OVER:

×

4005

  tFreeSMDropTbsReq(&dropReq);

×

4006

  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);

×

4007

  TAOS_RETURN(code);

×

4010

static int32_t createDropTbBatchReq(const SVDropTbReq *pReq, SVDropTbBatchReq *pBatchReq) {

×

4011

  pBatchReq->nReqs = 1;

×

4012

  pBatchReq->pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));

×

4013

  if (!pBatchReq->pArray) return terrno;

×

4014

  if (taosArrayPush(pBatchReq->pArray, pReq) == NULL) {

×

4015

    taosArrayDestroy(pBatchReq->pArray);

×

4016

    pBatchReq->pArray = NULL;

×

4017

    return terrno;

×

4019

  return TSDB_CODE_SUCCESS;

×

4022

static void destroySVDropTbBatchReqs(void *p) {

×

4023

  SVDropTbBatchReq *pReq = p;

×

4024

  taosArrayDestroy(pReq->pArray);

×

4025

  pReq->pArray = NULL;

×

4026

×

4028

static int32_t mndDropTbAdd(SMnode *pMnode, SHashObj *pVgHashMap, const SVgroupInfo *pVgInfo, char *name, tb_uid_t suid,

×

4030

  SVDropTbReq req = {.name = name, .suid = suid, .igNotExists = ignoreNotExists, .uid = 0};

×

4032

  SVDropTbVgReqs *pVgReqs = taosHashGet(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));

×

4033

  SVDropTbVgReqs  vgReqs = {0};

×

4034

  if (pVgReqs == NULL) {

×

4035

    vgReqs.info = *pVgInfo;

×

4036

    vgReqs.pBatchReqs = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbBatchReq));

×

4037

    if (!vgReqs.pBatchReqs) return terrno;

×

4038

    SVDropTbBatchReq batchReq = {0};

×

4039

    int32_t          code = createDropTbBatchReq(&req, &batchReq);

×

4040

    if (TSDB_CODE_SUCCESS != code) return code;

×

4041

    if (taosArrayPush(vgReqs.pBatchReqs, &batchReq) == NULL) {

×

4042

      taosArrayDestroy(batchReq.pArray);

×

4043

      return terrno;

×

4045

    if (taosHashPut(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId), &vgReqs, sizeof(vgReqs)) != 0) {

×

4046

      taosArrayDestroyEx(vgReqs.pBatchReqs, destroySVDropTbBatchReqs);

×

4047

      return terrno;

×

4050

    SVDropTbBatchReq batchReq = {0};

×

4051

    int32_t          code = createDropTbBatchReq(&req, &batchReq);

×

4052

    if (TSDB_CODE_SUCCESS != code) return code;

×

4053

    if (taosArrayPush(pVgReqs->pBatchReqs, &batchReq) == NULL) {

×

4054

      taosArrayDestroy(batchReq.pArray);

×

4055

      return terrno;

×

4058

  return 0;

×

4061

static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId) {

×

4062

  int32_t code = 0;

×

4064

  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);

×

4065

  if (!pVgObj) {

×

4066

    code = 0;

×

4067

    goto _end;

×

4069

  SVgroupInfo vgInfo = {.hashBegin = pVgObj->hashBegin,

×

4070

                        .hashEnd = pVgObj->hashEnd,

×

4071

                        .numOfTable = pVgObj->numOfTables,

×

4072

                        .vgId = pVgObj->vgId};

×

4073

  vgInfo.epSet = mndGetVgroupEpset(pMnode, pVgObj);

×

4074

  mndReleaseVgroup(pMnode, pVgObj);

×

4076

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

×

4077

    SVDropTbReq *pTb = taosArrayGet(pTbs, i);

×

4078

    TAOS_CHECK_GOTO(mndDropTbAdd(pMnode, pCtx->pVgMap, &vgInfo, pTb->name, pTb->suid, pTb->igNotExists), NULL, _end);

×

4080

_end:

×

4081

  return code;

×

4084

static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pRsp) {

×

4085

  int32_t                 code = -1;

×

4086

  SDecoder                decoder = {0};

×

4087

  SMnode                 *pMnode = pRsp->info.node;

×

4088

  SVFetchTtlExpiredTbsRsp rsp = {0};

×

4089

  SMndDropTbsWithTsmaCtx *pCtx = NULL;

×

4090

  if (pRsp->code != TSDB_CODE_SUCCESS) {

×

4091

    code = pRsp->code;

×

4092

    goto _end;

×

4094

  if (pRsp->contLen == 0) {

×

4095

    code = 0;

×

4096

    goto _end;

×

4099

  tDecoderInit(&decoder, pRsp->pCont, pRsp->contLen);

×

4100

  code = tDecodeVFetchTtlExpiredTbsRsp(&decoder, &rsp);

×

4101

  if (code) goto _end;

×

4103

  code = mndInitDropTbsWithTsmaCtx(&pCtx);

×

4104

  if (code) goto _end;

×

4106

  code = mndDropTbForSingleVg(pMnode, pCtx, rsp.pExpiredTbs, rsp.vgId);

×

4107

  if (code) goto _end;

×

4108

  code = mndCreateDropTbsTxnPrepare(pRsp, pCtx);

×

4109

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

4110

_end:

×

4111

  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);

×

4112

  tDecoderClear(&decoder);

×

4113

  tFreeFetchTtlExpiredTbsRsp(&rsp);

×

4114

  TAOS_RETURN(code);

×

taosdata / TDengine / #3633

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

taosdata / TDengine / #3633

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