taosdata/TDengine | Build 3798 | 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

  bool    hasTypeMod = false;

×

121

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

123

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

×

124

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

×

125

                 taosArrayGetSize(pStb->pFuncs) * TSDB_FUNC_NAME_LEN + sizeof(int32_t) * pStb->numOfColumns;

×

126

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

×

127

  if (pRaw == NULL) goto _OVER;

×

129

  int32_t dataPos = 0;

×

130

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

×

131

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

×

132

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

×

133

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

×

134

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

×

135

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

×

136

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

×

137

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

×

138

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

×

139

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

×

140

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

×

141

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

×

142

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

×

143

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

×

144

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

×

145

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

×

146

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

×

147

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

×

148

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

×

149

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

×

150

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

×

152

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

×

153

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

×

154

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

×

155

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

×

156

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

×

157

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

×

158

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

×

159

    hasTypeMod = hasTypeMod || HAS_TYPE_MOD(pSchema);

×

162

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

×

163

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

×

164

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

×

165

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

×

166

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

×

167

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

×

168

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

×

171

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

×

172

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

×

173

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

×

176

  if (pStb->commentLen > 0) {

×

177

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

×

180

  if (pStb->ast1Len > 0) {

×

181

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

×

184

  if (pStb->ast2Len > 0) {

×

185

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

×

188

  if (pStb->pCmpr != NULL) {

×

189

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

×

190

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

×

191

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

×

192

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

×

195

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

×

197

  if (hasTypeMod) {

×

198

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

×

199

      SDB_SET_INT32(pRaw, dataPos, pStb->pExtSchemas[i].typeMod, _OVER);

×

203

  SDB_SET_INT8(pRaw, dataPos, pStb->virtualStb, _OVER)

×

204

  SDB_SET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)

×

205

  SDB_SET_DATALEN(pRaw, dataPos, _OVER)

×

207

  terrno = 0;

×

209

_OVER:

×

210

  if (terrno != 0) {

×

211

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

×

212

    sdbFreeRaw(pRaw);

×

213

    return NULL;

×

216

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

×

217

  return pRaw;

×

220

static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {

×

221

  int32_t code = 0;

×

222

  int32_t lino = 0;

×

223

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

224

  SSdbRow *pRow = NULL;

×

225

  SStbObj *pStb = NULL;

×

226

  bool     hasExtSchemas = false;

×

228

  int8_t sver = 0;

×

229

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

×

231

  if (sver > STB_VER_NUMBER) {

×

232

    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;

×

233

    goto _OVER;

×

236

  pRow = sdbAllocRow(sizeof(SStbObj));

×

237

  if (pRow == NULL) goto _OVER;

×

239

  pStb = sdbGetRowObj(pRow);

×

240

  if (pStb == NULL) goto _OVER;

×

242

  int32_t dataPos = 0;

×

243

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

×

244

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

×

245

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

×

246

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

×

247

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

×

248

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

×

249

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

×

250

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

×

251

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

×

252

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

×

253

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

×

254

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

×

255

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

×

256

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

×

257

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

×

258

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

×

259

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

×

260

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

×

261

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

×

262

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

×

263

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

×

265

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

×

266

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

×

267

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

×

268

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

×

269

    goto _OVER;

×

272

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

×

273

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

×

274

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

×

275

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

×

276

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

×

277

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

×

278

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

×

279

    hasExtSchemas = hasExtSchemas || HAS_TYPE_MOD(pSchema);

×

282

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

×

283

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

×

284

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

×

285

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

×

286

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

×

287

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

×

288

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

×

291

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

×

292

    char funcName[TSDB_FUNC_NAME_LEN] = {0};

×

293

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

×

294

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

×

297

  if (pStb->commentLen > 0) {

×

298

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

×

299

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

×

300

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

×

303

  if (pStb->ast1Len > 0) {

×

304

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

×

305

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

×

306

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

×

309

  if (pStb->ast2Len > 0) {

×

310

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

×

311

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

×

312

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

×

315

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

×

316

  if (sver < STB_VER_NUMBER - 1) {

×

319

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

×

320

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

×

321

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

×

322

      pCmpr->id = pSchema->colId;

×

323

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

×

326

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

×

327

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

×

328

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

×

329

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

×

332

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

×

335

  if (hasExtSchemas) {

×

336

    pStb->pExtSchemas = taosMemoryCalloc(pStb->numOfColumns, sizeof(SExtSchema));

×

337

    if (!pStb->pExtSchemas) goto _OVER;

×

338

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

×

339

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

×

340

      SDB_GET_INT32(pRaw, dataPos, &pStb->pExtSchemas[i].typeMod, _OVER)

×

344

  if (sver < STB_VER_NUMBER) {

×

345

    pStb->virtualStb = 0;

×

347

    SDB_GET_INT8(pRaw, dataPos, &pStb->virtualStb, _OVER)

×

350

  SDB_GET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)

×

352

  terrno = 0;

×

354

_OVER:

×

355

  if (terrno != 0) {

×

356

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

×

357

    if (pStb != NULL) {

×

358

      taosMemoryFreeClear(pStb->pColumns);

×

359

      taosMemoryFreeClear(pStb->pTags);

×

360

      taosMemoryFreeClear(pStb->comment);

×

361

      taosMemoryFree(pStb->pCmpr);

×

362

      taosMemoryFreeClear(pStb->pExtSchemas);

×

364

    taosMemoryFreeClear(pRow);

×

365

    return NULL;

×

368

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

×

369

  return pRow;

×

372

void mndFreeStb(SStbObj *pStb) {

×

373

  taosArrayDestroy(pStb->pFuncs);

×

374

  taosMemoryFreeClear(pStb->pColumns);

×

375

  taosMemoryFreeClear(pStb->pTags);

×

376

  taosMemoryFreeClear(pStb->comment);

×

377

  taosMemoryFreeClear(pStb->pAst1);

×

378

  taosMemoryFreeClear(pStb->pAst2);

×

379

  taosMemoryFreeClear(pStb->pCmpr);

×

380

  taosMemoryFreeClear(pStb->pExtSchemas);

×

381

×

383

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

×

384

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

×

385

  return 0;

×

388

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

×

389

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

×

390

  mndFreeStb(pStb);

×

391

  return 0;

×

394

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

×

395

  terrno = 0;

×

396

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

×

398

  taosWLockLatch(&pOld->lock);

×

399

  int32_t numOfColumns = pOld->numOfColumns;

×

400

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

×

401

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

×

402

    if (pColumns == NULL) {

×

403

      goto END;

×

405

    taosMemoryFree(pOld->pColumns);

×

406

    pOld->pColumns = pColumns;

×

409

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

×

410

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

×

411

    if (pTags == NULL) {

×

412

      goto END;

×

414

    taosMemoryFree(pOld->pTags);

×

415

    pOld->pTags = pTags;

×

418

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

×

419

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

×

420

    if (comment == NULL) {

×

421

      goto END;

×

423

    taosMemoryFree(pOld->comment);

×

424

    pOld->comment = comment;

×

426

  pOld->commentLen = pNew->commentLen;

×

428

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

×

429

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

×

430

    if (pAst1 == NULL) {

×

431

      goto END;

×

433

    taosMemoryFree(pOld->pAst1);

×

434

    pOld->pAst1 = pAst1;

×

437

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

×

438

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

×

439

    if (pAst2 == NULL) {

×

440

      goto END;

×

442

    taosMemoryFree(pOld->pAst2);

×

443

    pOld->pAst2 = pAst2;

×

446

  pOld->updateTime = pNew->updateTime;

×

447

  pOld->tagVer = pNew->tagVer;

×

448

  pOld->colVer = pNew->colVer;

×

449

  pOld->smaVer = pNew->smaVer;

×

450

  pOld->nextColId = pNew->nextColId;

×

451

  pOld->ttl = pNew->ttl;

×

452

  pOld->keep = pNew->keep;

×

454

  if (pNew->numOfColumns > 0) {

×

455

    pOld->numOfColumns = pNew->numOfColumns;

×

456

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

×

458

  if (pNew->numOfTags > 0) {

×

459

    pOld->numOfTags = pNew->numOfTags;

×

460

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

×

462

  if (pNew->commentLen > 0) {

×

463

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

×

464

    pOld->commentLen = pNew->commentLen;

×

466

  if (pNew->ast1Len != 0) {

×

467

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

×

468

    pOld->ast1Len = pNew->ast1Len;

×

470

  if (pNew->ast2Len != 0) {

×

471

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

×

472

    pOld->ast2Len = pNew->ast2Len;

×

474

  if (numOfColumns < pNew->numOfColumns) {

×

475

    taosMemoryFree(pOld->pCmpr);

×

476

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

×

477

    if (pOld->pCmpr == NULL){

×

478

      goto END;

×

480

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

×

482

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

×

485

  if (pNew->pExtSchemas) {

×

486

    taosMemoryFreeClear(pOld->pExtSchemas);

×

487

    pOld->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));

×

488

    if (pOld->pExtSchemas == NULL){

×

489

      goto END;

×

491

    memcpy(pOld->pExtSchemas, pNew->pExtSchemas, pNew->numOfColumns * sizeof(SExtSchema));

×

494

END:

×

495

  taosWUnLockLatch(&pOld->lock);

×

496

  return terrno;

×

499

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

×

500

  SSdb    *pSdb = pMnode->pSdb;

×

501

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

×

502

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

×

503

    terrno = TSDB_CODE_MND_STB_NOT_EXIST;

×

505

  return pStb;

×

508

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

×

509

  SSdb *pSdb = pMnode->pSdb;

×

510

  sdbRelease(pSdb, pStb);

×

511

×

513

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

×

514

  SName name = {0};

×

515

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

×

517

  char db[TSDB_TABLE_FNAME_LEN] = {0};

×

518

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

×

520

  return mndAcquireDb(pMnode, db);

×

532

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

×

534

  SEncoder       encoder = {0};

×

536

  SName          name = {0};

×

537

  SVCreateStbReq req = {0};

×

539

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

×

540

    goto _err;

×

542

  char dbFName[TSDB_DB_FNAME_LEN] = {0};

×

543

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

×

544

    goto _err;

×

547

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

×

548

  req.suid = pStb->uid;

×

549

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

×

550

  req.alterOriData = alterOriData;

×

551

  req.alterOriDataLen = alterOriDataLen;

×

552

  req.source = pStb->source;

×

553

  req.virtualStb = pStb->virtualStb;

×

555

  req.schemaRow.nCols = pStb->numOfColumns;

×

556

  req.schemaRow.version = pStb->colVer;

×

557

  req.schemaRow.pSchema = pStb->pColumns;

×

558

  req.schemaTag.nCols = pStb->numOfTags;

×

559

  req.schemaTag.version = pStb->tagVer;

×

560

  req.schemaTag.pSchema = pStb->pTags;

×

562

  req.colCmpred = 1;

×

563

  SColCmprWrapper *pCmpr = &req.colCmpr;

×

564

  req.keep = pStb->keep;

×

565

  pCmpr->version = pStb->colVer;

×

566

  pCmpr->nCols = pStb->numOfColumns;

×

568

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

×

569

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

×

570

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

×

571

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

×

572

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

×

575

  if (req.rollup) {

×

576

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

×

577

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

×

578

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

×

579

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

×

580

    if (pStb->ast1Len > 0) {

×

581

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

×

584

        goto _err;

×

587

    if (pStb->ast2Len > 0) {

×

588

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

×

591

        goto _err;

×

595

  req.pExtSchemas = pStb->pExtSchemas; // only reference to it.

×

597

  int32_t ret = 0;

×

598

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

×

599

  if (ret < 0) {

×

600

    goto _err;

×

603

  contLen += sizeof(SMsgHead);

×

605

  SMsgHead *pHead = taosMemoryCalloc(1, contLen);

×

606

  if (pHead == NULL) {

×

607

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

608

    goto _err;

×

611

  pHead->contLen = htonl(contLen);

×

612

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

×

614

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

×

615

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

×

616

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

×

617

    taosMemoryFreeClear(pHead);

×

618

    tEncoderClear(&encoder);

×

619

    goto _err;

×

621

  tEncoderClear(&encoder);

×

623

  *pContLen = contLen;

×

624

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

×

625

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

×

626

  taosMemoryFreeClear(req.colCmpr.pColCmpr);

×

627

  return pHead;

×

628

_err:

×

629

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

×

630

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

×

631

  taosMemoryFreeClear(req.colCmpr.pColCmpr);

×

632

  return NULL;

×

635

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

×

636

  SName        name = {0};

×

637

  SVDropStbReq req = {0};

×

638

  int32_t      contLen = 0;

×

639

  int32_t      ret = 0;

×

640

  SMsgHead    *pHead = NULL;

×

641

  SEncoder     encoder = {0};

×

643

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

×

644

    return NULL;

×

647

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

×

648

  req.suid = pStb->uid;

×

650

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

×

651

  if (ret < 0) return NULL;

×

653

  contLen += sizeof(SMsgHead);

×

654

  pHead = taosMemoryMalloc(contLen);

×

655

  if (pHead == NULL) {

×

656

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

657

    return NULL;

×

660

  pHead->contLen = htonl(contLen);

×

661

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

×

663

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

×

665

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

×

666

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

×

667

  tEncoderClear(&encoder);

×

668

  if (code != 0) {

×

669

    terrno = code;

×

670

    return NULL;

×

673

  *pContLen = contLen;

×

674

  return pHead;

×

677

int32_t mndCheckCreateStbReq(SMCreateStbReq *pCreate) {

×

678

  int32_t code = 0;

×

679

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

×

680

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

681

    TAOS_RETURN(code);

×

684

  if (pCreate->virtualStb != 0 && pCreate->virtualStb != 1) {

×

685

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

686

    TAOS_RETURN(code);

×

689

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

×

690

    code = TSDB_CODE_PAR_INVALID_COLUMNS_NUM;

×

691

    TAOS_RETURN(code);

×

694

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

×

695

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

696

    TAOS_RETURN(code);

×

699

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

×

700

  if (pField->type != TSDB_DATA_TYPE_TIMESTAMP) {

×

701

    code = TSDB_CODE_PAR_INVALID_FIRST_COLUMN;

×

702

    TAOS_RETURN(code);

×

705

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

×

706

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

×

707

    if (pField1->type >= TSDB_DATA_TYPE_MAX) {

×

708

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

709

      TAOS_RETURN(code);

×

711

    if (pField1->bytes <= 0) {

×

712

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

713

      TAOS_RETURN(code);

×

715

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

×

716

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

717

      TAOS_RETURN(code);

×

721

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

×

722

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

×

723

    if (pField1->type >= TSDB_DATA_TYPE_MAX) {

×

724

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

725

      TAOS_RETURN(code);

×

727

    if (pField1->bytes <= 0) {

×

728

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

729

      TAOS_RETURN(code);

×

731

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

×

732

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

733

      TAOS_RETURN(code);

×

737

  TAOS_RETURN(code);

×

740

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

×

741

  int32_t  code = 0;

×

742

  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);

×

743

  if (pRedoRaw == NULL) {

×

744

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

745

    if (terrno != 0) code = terrno;

×

746

    TAOS_RETURN(code);

×

748

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

×

749

    sdbFreeRaw(pRedoRaw);

×

750

    TAOS_RETURN(code);

×

752

  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));

×

754

  TAOS_RETURN(code);

×

757

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

×

758

  int32_t  code = 0;

×

759

  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);

×

760

  if (pCommitRaw == NULL) {

×

761

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

762

    if (terrno != 0) code = terrno;

×

763

    TAOS_RETURN(code);

×

765

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

×

766

    sdbFreeRaw(pCommitRaw);

×

767

    TAOS_RETURN(code);

×

769

  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

×

771

  TAOS_RETURN(code);

×

774

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

×

775

  int32_t code = 0;

×

776

  SSdb   *pSdb = pMnode->pSdb;

×

777

  SVgObj *pVgroup = NULL;

×

778

  void   *pIter = NULL;

×

781

  while (1) {

×

782

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

×

783

    if (pIter == NULL) break;

×

784

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

×

785

      sdbRelease(pSdb, pVgroup);

×

786

      continue;

×

789

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

×

790

    if (pReq == NULL) {

×

791

      sdbCancelFetch(pSdb, pIter);

×

792

      sdbRelease(pSdb, pVgroup);

×

793

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

794

      if (terrno != 0) code = terrno;

×

795

      TAOS_RETURN(code);

×

798

    STransAction action = {0};

×

799

    action.mTraceId = pTrans->mTraceId;

×

800

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

801

    action.pCont = pReq;

×

802

    action.contLen = contLen;

×

803

    action.msgType = TDMT_VND_CREATE_STB;

×

804

    action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;

×

805

    action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

806

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

×

807

      taosMemoryFree(pReq);

×

808

      sdbCancelFetch(pSdb, pIter);

×

809

      sdbRelease(pSdb, pVgroup);

×

810

      TAOS_RETURN(code);

×

812

    sdbRelease(pSdb, pVgroup);

×

815

  TAOS_RETURN(code);

×

818

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

×

819

  int32_t code = 0;

×

820

  SSdb   *pSdb = pMnode->pSdb;

×

823

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

×

824

  if (pReq == NULL) {

×

825

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

826

    if (terrno != 0) code = terrno;

×

827

    TAOS_RETURN(code);

×

830

  STransAction action = {0};

×

831

  action.mTraceId = pTrans->mTraceId;

×

832

  action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

833

  action.pCont = pReq;

×

834

  action.contLen = contLen;

×

835

  action.msgType = TDMT_VND_CREATE_STB;

×

836

  action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;

×

837

  action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

838

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

×

839

    taosMemoryFree(pReq);

×

840

    TAOS_RETURN(code);

×

843

  TAOS_RETURN(code);

×

846

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

×

847

  int32_t code = 0;

×

848

  SSdb   *pSdb = pMnode->pSdb;

×

849

  SVgObj *pVgroup = NULL;

×

850

  void   *pIter = NULL;

×

852

  while (1) {

×

853

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

×

854

    if (pIter == NULL) break;

×

855

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

×

856

      sdbRelease(pSdb, pVgroup);

×

857

      continue;

×

860

    int32_t contLen = 0;

×

861

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

×

862

    if (pReq == NULL) {

×

863

      sdbCancelFetch(pSdb, pIter);

×

864

      sdbRelease(pSdb, pVgroup);

×

865

      code = TSDB_CODE_OUT_OF_MEMORY;

×

866

      TAOS_RETURN(code);

×

869

    STransAction action = {0};

×

870

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

871

    action.pCont = pReq;

×

872

    action.contLen = contLen;

×

873

    action.msgType = TDMT_VND_DROP_STB;

×

874

    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

875

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

×

876

      taosMemoryFree(pReq);

×

877

      sdbCancelFetch(pSdb, pIter);

×

878

      sdbRelease(pSdb, pVgroup);

×

879

      TAOS_RETURN(code);

×

881

    sdbRelease(pSdb, pVgroup);

×

884

  TAOS_RETURN(code);

×

887

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

×

888

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

×

889

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

×

890

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

×

891

      return pSchema;

×

894

  return NULL;

×

897

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

×

898

  int32_t code = 0;

×

899

  bool    hasTypeMods = false;

×

900

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

×

901

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

×

902

  pDst->createdTime = taosGetTimestampMs();

×

903

  pDst->updateTime = pDst->createdTime;

×

904

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

×

906

                  : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);

×

907

  pDst->dbUid = pDb->uid;

×

908

  pDst->tagVer = 1;

×

909

  pDst->colVer = 1;

×

910

  pDst->smaVer = 1;

×

911

  pDst->nextColId = 1;

×

912

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

×

913

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

×

914

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

×

915

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

×

916

  pDst->ttl = pCreate->ttl;

×

917

  pDst->numOfColumns = pCreate->numOfColumns;

×

918

  pDst->numOfTags = pCreate->numOfTags;

×

919

  pDst->numOfFuncs = pCreate->numOfFuncs;

×

920

  pDst->commentLen = pCreate->commentLen;

×

921

  pDst->pFuncs = pCreate->pFuncs;

×

922

  pDst->source = pCreate->source;

×

923

  pDst->keep = pCreate->keep;

×

924

  pDst->virtualStb = pCreate->virtualStb;

×

925

  pCreate->pFuncs = NULL;

×

927

  if (pDst->commentLen > 0) {

×

928

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

×

929

    if (pDst->comment == NULL) {

×

930

      code = terrno;

×

931

      TAOS_RETURN(code);

×

933

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

×

936

  pDst->ast1Len = pCreate->ast1Len;

×

937

  if (pDst->ast1Len > 0) {

×

938

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

×

939

    if (pDst->pAst1 == NULL) {

×

940

      code = terrno;

×

941

      TAOS_RETURN(code);

×

943

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

×

946

  pDst->ast2Len = pCreate->ast2Len;

×

947

  if (pDst->ast2Len > 0) {

×

948

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

×

949

    if (pDst->pAst2 == NULL) {

×

950

      code = terrno;

×

951

      TAOS_RETURN(code);

×

953

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

×

956

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

×

957

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

×

958

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

×

959

    code = terrno;

×

960

    TAOS_RETURN(code);

×

963

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

×

964

    code = TSDB_CODE_OUT_OF_RANGE;

×

965

    TAOS_RETURN(code);

×

968

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

×

969

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

×

970

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

×

971

    pSchema->type = pField->type;

×

972

    pSchema->bytes = pField->bytes;

×

973

    pSchema->flags = pField->flags;

×

974

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

×

975

    pSchema->colId = pDst->nextColId;

×

976

    pDst->nextColId++;

×

977

    hasTypeMods = hasTypeMods || HAS_TYPE_MOD(pSchema);

×

980

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

×

981

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

×

982

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

×

983

    pSchema->type = pField->type;

×

984

    pSchema->bytes = pField->bytes;

×

985

    if (i == 0) {

×

986

      SSCHMEA_SET_IDX_ON(pSchema);

×

988

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

×

989

    pSchema->colId = pDst->nextColId;

×

990

    pDst->nextColId++;

×

993

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

×

994

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

×

995

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

×

996

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

×

998

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

×

999

    pColCmpr->id = pSchema->colId;

×

1000

    pColCmpr->alg = pField->compress;

×

1003

  if (hasTypeMods) {

×

1004

    pDst->pExtSchemas = taosMemoryCalloc(pDst->numOfColumns, sizeof(SExtSchema));

×

1005

    if (!pDst->pExtSchemas) {

×

1006

      code = terrno;

×

1007

      TAOS_RETURN(code);

×

1009

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

×

1010

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

×

1011

      pDst->pExtSchemas[i].typeMod = pField->typeMod;

×

1014

  TAOS_RETURN(code);

×

1016

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

×

1017

  SName name = {0};

×

1018

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

×

1019

    return -1;

×

1021

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

×

1024

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

×

1025

  SStbObj stbObj = {0};

×

1026

  int32_t code = -1;

×

1028

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

×

1030

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

×

1031

  if (pTrans == NULL) {

×

1032

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

1033

    if (terrno != 0) code = terrno;

×

1034

    goto _OVER;

×

1037

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

×

1038

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

×

1040

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

×

1041

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

×

1042

    goto _OVER;

×

1044

  SSIdx idx = {0};

×

1045

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

×

1046

    code = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;

×

1047

    mndReleaseIdx(pMnode, idx.pIdx);

×

1048

    goto _OVER;

×

1051

  SIdxObj idxObj = {0};

×

1052

  memcpy(idxObj.name, fullIdxName, TSDB_INDEX_FNAME_LEN);

×

1053

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

×

1054

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

×

1055

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

×

1056

  idxObj.createdTime = taosGetTimestampMs();

×

1057

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

×

1058

  idxObj.stbUid = stbObj.uid;

×

1059

  idxObj.dbUid = stbObj.dbUid;

×

1061

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

×

1063

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

×

1064

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

×

1065

  code = 0;

×

1067

_OVER:

×

1068

  mndTransDrop(pTrans);

×

1069

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

×

1070

  TAOS_RETURN(code);

×

1073

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

×

1074

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

×

1075

  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));

×

1076

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

×

1077

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

×

1078

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

×

1079

  return 0;

×

1082

static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {

×

1083

  SMnode           *pMnode = pReq->info.node;

×

1084

  SSdb             *pSdb = pMnode->pSdb;

×

1085

  SVgObj           *pVgroup = NULL;

×

1086

  void             *pIter = NULL;

×

1087

  SVDropTtlTableReq ttlReq = {

×

1088

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

×

1089

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

×

1090

  int32_t contLen = reqLen + sizeof(SMsgHead);

×

1092

  mDebug("start to process ttl timer");

×

1094

  while (1) {

×

1095

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

×

1096

    if (pIter == NULL) break;

×

1098

    int32_t   code = 0;

×

1099

    SMsgHead *pHead = rpcMallocCont(contLen);

×

1100

    if (pHead == NULL) {

×

1101

      sdbRelease(pSdb, pVgroup);

×

1102

      continue;

×

1104

    pHead->contLen = htonl(contLen);

×

1105

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

×

1106

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

×

1107

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

×

1108

      sdbRelease(pSdb, pVgroup);

×

1109

      continue;

×

1112

    SRpcMsg rpcMsg = {

×

1114

    SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

1115

    code = tmsgSendReq(&epSet, &rpcMsg);

×

1116

    if (code != 0) {

×

1117

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

×

1119

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

×

1121

    sdbRelease(pSdb, pVgroup);

×

1124

  return 0;

×

1127

static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {

×

1128

  SMnode     *pMnode = pReq->info.node;

×

1129

  SSdb       *pSdb = pMnode->pSdb;

×

1130

  SVgObj     *pVgroup = NULL;

×

1131

  void       *pIter = NULL;

×

1132

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

×

1133

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

×

1134

  int32_t     contLen = reqLen + sizeof(SMsgHead);

×

1136

  while (1) {

×

1137

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

×

1138

    if (pIter == NULL) break;

×

1140

    int32_t code = 0;

×

1142

    SMsgHead *pHead = rpcMallocCont(contLen);

×

1143

    if (pHead == NULL) {

×

1144

      sdbCancelFetch(pSdb, pVgroup);

×

1145

      sdbRelease(pSdb, pVgroup);

×

1146

      continue;

×

1148

    pHead->contLen = htonl(contLen);

×

1149

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

×

1150

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

×

1151

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

×

1154

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

×

1155

    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

1156

    code = tmsgSendReq(&epSet, &rpcMsg);

×

1157

    if (code != 0) {

×

1158

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

×

1160

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

×

1162

    sdbRelease(pSdb, pVgroup);

×

1165

  return 0;

×

1168

static int32_t mndProcessS3MigrateDbTimer(SRpcMsg *pReq) {

×

1169

  SMnode          *pMnode = pReq->info.node;

×

1170

  SSdb            *pSdb = pMnode->pSdb;

×

1171

  SVgObj          *pVgroup = NULL;

×

1172

  void            *pIter = NULL;

×

1173

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

×

1174

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

×

1175

  int32_t          contLen = reqLen + sizeof(SMsgHead);

×

1177

  while (1) {

×

1178

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

×

1179

    if (pIter == NULL) break;

×

1181

    int32_t code = 0;

×

1183

    SMsgHead *pHead = rpcMallocCont(contLen);

×

1184

    if (pHead == NULL) {

×

1185

      sdbRelease(pSdb, pVgroup);

×

1186

      continue;

×

1188

    pHead->contLen = htonl(contLen);

×

1189

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

×

1190

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

×

1191

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

×

1192

      sdbRelease(pSdb, pVgroup);

×

1193

      continue;

×

1196

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

×

1197

    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

1198

    code = tmsgSendReq(&epSet, &rpcMsg);

×

1199

    if (code != 0) {

×

1200

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

×

1202

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

×

1204

    sdbRelease(pSdb, pVgroup);

×

1207

  return 0;

×

1210

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

×

1211

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

×

1212

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

×

1213

      return tag;

×

1217

  return -1;

×

1220

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

×

1221

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

×

1222

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

×

1223

      return col;

×

1227

  return -1;

×

1230

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

×

1231

  int32_t rowLen = 0;

×

1232

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

×

1233

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

×

1236

  int32_t nField = taosArrayGetSize(pFields);

×

1237

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

×

1238

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

×

1241

  return rowLen <= maxLen;

×

1244

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

×

1245

  int32_t code = 0;

×

1246

  taosRLockLatch(&pStb->lock);

×

1247

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

×

1248

  taosRUnLockLatch(&pStb->lock);

×

1250

  pDst->source = createReq->source;

×

1251

  pDst->updateTime = taosGetTimestampMs();

×

1252

  pDst->numOfColumns = createReq->numOfColumns;

×

1253

  pDst->numOfTags = createReq->numOfTags;

×

1254

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

×

1255

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

×

1256

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

×

1257

  pDst->pExtSchemas = taosMemoryCalloc(pDst->numOfColumns, sizeof(SExtSchema));

×

1259

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

×

1260

    code = terrno;

×

1261

    TAOS_RETURN(code);

×

1264

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

×

1265

    code = TSDB_CODE_OUT_OF_RANGE;

×

1266

    TAOS_RETURN(code);

×

1269

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

×

1270

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

×

1271

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

×

1272

    pSchema->type = pField->type;

×

1273

    pSchema->bytes = pField->bytes;

×

1274

    pSchema->flags = pField->flags;

×

1275

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

×

1276

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

×

1277

    if (cIndex >= 0) {

×

1278

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

×

1280

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

×

1284

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

×

1285

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

×

1286

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

×

1287

    pSchema->type = pField->type;

×

1288

    pSchema->bytes = pField->bytes;

×

1289

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

×

1290

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

×

1291

    if (cIndex >= 0) {

×

1292

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

×

1294

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

×

1297

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

×

1298

    SColCmpr          *p = pDst->pCmpr + i;

×

1299

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

×

1300

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

×

1301

    p->id = pSchema->colId;

×

1302

    if (pField->compress == 0) {

×

1303

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

×

1305

      p->alg = pField->compress;

×

1307

    if (pField->flags & COL_HAS_TYPE_MOD) {

×

1308

      pDst->pExtSchemas[i].typeMod = pField->typeMod;

×

1311

  pDst->tagVer = createReq->tagVer;

×

1312

  pDst->colVer = createReq->colVer;

×

1313

  return TSDB_CODE_SUCCESS;

×

1316

static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {

×

1317

  SMnode        *pMnode = pReq->info.node;

×

1318

  int32_t        code = -1;

×

1319

  SStbObj       *pStb = NULL;

×

1320

  SDbObj        *pDb = NULL;

×

1321

  SMCreateStbReq createReq = {0};

×

1322

  bool           isAlter = false;

×

1323

  SHashObj      *pHash = NULL;

×

1325

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

×

1326

    code = TSDB_CODE_INVALID_MSG;

×

1327

    goto _OVER;

×

1330

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

×

1331

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

×

1332

    code = TSDB_CODE_INVALID_MSG;

×

1333

    goto _OVER;

×

1336

  pStb = mndAcquireStb(pMnode, createReq.name);

×

1337

  if (pStb != NULL) {

×

1338

    if (createReq.igExists) {

×

1339

      if (createReq.source == TD_REQ_FROM_APP) {

×

1340

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

×

1341

        code = 0;

×

1342

        goto _OVER;

×

1343

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

×

1344

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

×

1345

        code = 0;

×

1346

        goto _OVER;

×

1347

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

×

1348

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

×

1349

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

×

1350

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

×

1352

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

×

1353

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

×

1354

          code = 0;

×

1355

          goto _OVER;

×

1356

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

×

1357

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

×

1358

          isAlter = true;

×

1359

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

×

1361

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

×

1362

          code = TSDB_CODE_MND_INVALID_SCHEMA_VER;

×

1363

          goto _OVER;

×

1366

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

×

1368

        code = TSDB_CODE_MND_INVALID_SCHEMA_VER;

×

1369

        goto _OVER;

×

1372

      code = TSDB_CODE_MND_STB_ALREADY_EXIST;

×

1373

      goto _OVER;

×

1375

  } else if (terrno != TSDB_CODE_MND_STB_NOT_EXIST) {

×

1376

    goto _OVER;

×

1377

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

×

1378

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

×

1379

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

×

1380

    code = 0;

×

1381

    goto _OVER;

×

1384

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

×

1386

  if (pHash == NULL) {

×

1387

    code = TSDB_CODE_OUT_OF_MEMORY;

×

1388

    goto _OVER;

×

1391

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

×

1392

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

×

1393

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

×

1394

      if (code == TSDB_CODE_DUP_KEY) {

×

1395

        code = TSDB_CODE_TSC_DUP_COL_NAMES;

×

1397

      goto _OVER;

×

1401

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

×

1402

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

×

1403

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

×

1404

      if (code == TSDB_CODE_DUP_KEY) {

×

1405

        code = TSDB_CODE_TSC_DUP_COL_NAMES;

×

1407

      goto _OVER;

×

1411

  pDb = mndAcquireDbByStb(pMnode, createReq.name);

×

1412

  if (pDb == NULL) {

×

1413

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

1414

    goto _OVER;

×

1417

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

×

1418

    goto _OVER;

×

1421

  int32_t numOfStbs = -1;

×

1422

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

×

1423

    goto _OVER;

×

1426

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

×

1427

    code = TSDB_CODE_MND_SINGLE_STB_MODE_DB;

×

1428

    goto _OVER;

×

1431

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

×

1432

    goto _OVER;

×

1435

  if (isAlter) {

×

1436

    bool    needRsp = false;

×

1437

    SStbObj pDst = {0};

×

1438

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

×

1439

      taosMemoryFreeClear(pDst.pTags);

×

1440

      taosMemoryFreeClear(pDst.pColumns);

×

1441

      taosMemoryFreeClear(pDst.pCmpr);

×

1442

      taosMemoryFreeClear(pDst.pExtSchemas);

×

1443

      goto _OVER;

×

1446

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

×

1447

    taosMemoryFreeClear(pDst.pTags);

×

1448

    taosMemoryFreeClear(pDst.pColumns);

×

1449

    taosMemoryFreeClear(pDst.pCmpr);

×

1450

    taosMemoryFreeClear(pDst.pExtSchemas);

×

1452

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

×

1454

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

1456

  SName name = {0};

×

1457

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

×

1459

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

×

1460

    char detail[1000] = {0};

×

1462

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

×

1464

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

×

1466

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

×

1468

_OVER:

×

1469

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

×

1470

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

×

1473

  mndReleaseStb(pMnode, pStb);

×

1474

  mndReleaseDb(pMnode, pDb);

×

1475

  tFreeSMCreateStbReq(&createReq);

×

1477

  if (pHash != NULL) {

×

1478

    taosHashCleanup(pHash);

×

1481

  TAOS_RETURN(code);

×

1484

static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) {

×

1485

  int32_t code = 0;

×

1486

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

×

1487

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

×

1488

  if (pAlter->keep != -1) return 0;

×

1490

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

×

1491

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1492

    TAOS_RETURN(code);

×

1495

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

×

1496

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

×

1497

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

×

1498

      code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1499

      TAOS_RETURN(code);

×

1503

  TAOS_RETURN(code);

×

1506

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

×

1507

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

×

1508

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

×

1509

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

×

1510

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

×

1511

    TAOS_RETURN(terrno);

×

1514

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

×

1515

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

×

1516

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

×

1517

  if (pOld->pExtSchemas) {

×

1518

    pNew->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));

×

1519

    if (pNew->pExtSchemas == NULL) {

×

1520

      TAOS_RETURN(terrno);

×

1522

    memcpy(pNew->pExtSchemas, pOld->pExtSchemas, sizeof(SExtSchema) * pOld->numOfColumns);

×

1525

  TAOS_RETURN(0);

×

1528

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

×

1530

  int32_t code = 0;

×

1531

  if (commentLen > 0) {

×

1532

    pNew->commentLen = commentLen;

×

1533

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

×

1534

    if (pNew->comment == NULL) {

×

1535

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1536

      return -1;

×

1538

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

×

1539

  } else if (commentLen == 0) {

×

1540

    pNew->commentLen = 0;

×

1544

  if (ttl >= 0) {

×

1545

    pNew->ttl = ttl;

×

1548

  if (keep > 0) {

×

1549

    pNew->keep = keep;

×

1552

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

×

1553

    TAOS_RETURN(code);

×

1555

  TAOS_RETURN(code);

×

1558

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

×

1559

  int32_t code = 0;

×

1560

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

×

1561

    code = TSDB_CODE_MND_TOO_MANY_TAGS;

×

1562

    TAOS_RETURN(code);

×

1565

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

×

1566

    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;

×

1567

    TAOS_RETURN(code);

×

1570

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

×

1571

    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;

×

1572

    TAOS_RETURN(code);

×

1575

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

×

1576

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

×

1577

    TAOS_RETURN(code);

×

1580

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

×

1581

    code = TSDB_CODE_OUT_OF_RANGE;

×

1582

    TAOS_RETURN(code);

×

1585

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

×

1586

    SField *pField = taosArrayGet(pFields, i);

×

1587

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

×

1588

      code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1589

      TAOS_RETURN(code);

×

1592

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

×

1593

      code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1594

      TAOS_RETURN(code);

×

1597

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

×

1598

    pSchema->bytes = pField->bytes;

×

1599

    pSchema->type = pField->type;

×

1600

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

×

1601

    pSchema->colId = pNew->nextColId;

×

1602

    pNew->nextColId++;

×

1604

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

×

1607

  pNew->tagVer++;

×

1608

  TAOS_RETURN(code);

×

1611

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

×

1612

  int32_t code = 0;

×

1613

  SSdb   *pSdb = pMnode->pSdb;

×

1614

  void   *pIter = NULL;

×

1615

  while (1) {

×

1616

    SMqTopicObj *pTopic = NULL;

×

1617

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

×

1618

    if (pIter == NULL) break;

×

1620

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

×

1622

    if (pTopic->ast == NULL) {

×

1623

      sdbRelease(pSdb, pTopic);

×

1624

      continue;

×

1627

    SNode *pAst = NULL;

×

1628

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

×

1629

      code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;

×

1630

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

×

1631

      sdbRelease(pSdb, pTopic);

×

1632

      sdbCancelFetch(pSdb, pIter);

×

1633

      TAOS_RETURN(code);

×

1636

    SNodeList *pNodeList = NULL;

×

1637

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

×

1639

      sdbRelease(pSdb, pTopic);

×

1640

      sdbCancelFetch(pSdb, pIter);

×

1641

      TAOS_RETURN(code);

×

1643

    SNode *pNode = NULL;

×

1644

    FOREACH(pNode, pNodeList) {

×

1645

      SColumnNode *pCol = (SColumnNode *)pNode;

×

1646

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

×

1649

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

×

1650

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

×

1651

        goto NEXT;

×

1653

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

×

1654

        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;

×

1655

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

×

1656

        nodesDestroyNode(pAst);

×

1657

        nodesDestroyList(pNodeList);

×

1658

        sdbCancelFetch(pSdb, pIter);

×

1659

        sdbRelease(pSdb, pTopic);

×

1660

        TAOS_RETURN(code);

×

1662

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

×

1665

  NEXT:

×

1666

    sdbRelease(pSdb, pTopic);

×

1667

    nodesDestroyNode(pAst);

×

1668

    nodesDestroyList(pNodeList);

×

1670

  TAOS_RETURN(code);

×

1673

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

×

1674

  int32_t code = 0;

×

1675

  SSdb   *pSdb = pMnode->pSdb;

×

1676

  void   *pIter = NULL;

×

1677

  while (1) {

×

1678

    SStreamObj *pStream = NULL;

×

1679

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

×

1680

    if (pIter == NULL) break;

×

1682

    SNode *pAst = NULL;

×

1683

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

×

1684

      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;

×

1685

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

×

1686

      sdbRelease(pSdb, pStream);

×

1687

      sdbCancelFetch(pSdb, pIter);

×

1688

      TAOS_RETURN(code);

×

1691

    SNodeList *pNodeList = NULL;

×

1692

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

×

1694

      sdbRelease(pSdb, pStream);

×

1695

      sdbCancelFetch(pSdb, pIter);

×

1696

      TAOS_RETURN(code);

×

1699

    SNode *pNode = NULL;

×

1700

    FOREACH(pNode, pNodeList) {

×

1701

      SColumnNode *pCol = (SColumnNode *)pNode;

×

1703

      if (pCol->tableId != suid) {

×

1704

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

×

1705

        goto NEXT;

×

1707

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

×

1708

        code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;

×

1709

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

×

1710

        nodesDestroyNode(pAst);

×

1711

        nodesDestroyList(pNodeList);

×

1712

        sdbRelease(pSdb, pStream);

×

1713

        sdbCancelFetch(pSdb, pIter);

×

1714

        TAOS_RETURN(code);

×

1716

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

×

1719

  NEXT:

×

1720

    sdbRelease(pSdb, pStream);

×

1721

    nodesDestroyNode(pAst);

×

1722

    nodesDestroyList(pNodeList);

×

1724

  TAOS_RETURN(code);

×

1727

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

×

1728

  int32_t code = 0;

×

1729

  SSdb   *pSdb = pMnode->pSdb;

×

1730

  void   *pIter = NULL;

×

1731

  while (1) {

×

1732

    SSmaObj *pSma = NULL;

×

1733

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

×

1734

    if (pIter == NULL) break;

×

1736

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

×

1739

    SNode *pAst = NULL;

×

1740

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

×

1741

      code = TSDB_CODE_SDB_INVALID_DATA_CONTENT;

×

1742

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

×

1744

      sdbCancelFetch(pSdb, pIter);

×

1745

      TAOS_RETURN(code);

×

1748

    SNodeList *pNodeList = NULL;

×

1749

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

×

1751

      sdbCancelFetch(pSdb, pIter);

×

1752

      TAOS_RETURN(code);

×

1754

    SNode *pNode = NULL;

×

1755

    FOREACH(pNode, pNodeList) {

×

1756

      SColumnNode *pCol = (SColumnNode *)pNode;

×

1757

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

×

1759

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

×

1760

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

×

1761

        goto NEXT;

×

1763

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

×

1764

        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;

×

1765

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

×

1766

        nodesDestroyNode(pAst);

×

1767

        nodesDestroyList(pNodeList);

×

1768

        sdbRelease(pSdb, pSma);

×

1769

        sdbCancelFetch(pSdb, pIter);

×

1770

        TAOS_RETURN(code);

×

1772

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

×

1775

  NEXT:

×

1776

    sdbRelease(pSdb, pSma);

×

1777

    nodesDestroyNode(pAst);

×

1778

    nodesDestroyList(pNodeList);

×

1780

  TAOS_RETURN(code);

×

1783

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

×

1784

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

×

1785

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

×

1786

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

×

1787

  TAOS_RETURN(0);

×

1790

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

×

1791

  int32_t code = 0;

×

1792

  int32_t tag = mndFindSuperTableTagIndex(pOld, tagName);

×

1793

  if (tag < 0) {

×

1794

    code = TSDB_CODE_MND_TAG_NOT_EXIST;

×

1795

    TAOS_RETURN(code);

×

1798

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

×

1799

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

×

1801

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1803

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

×

1804

  pNew->numOfTags--;

×

1806

  pNew->tagVer++;

×

1811

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

×

1812

  TAOS_RETURN(code);

×

1815

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

×

1816

  int32_t code = 0;

×

1817

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

×

1818

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1819

    TAOS_RETURN(code);

×

1822

  SField *pField0 = taosArrayGet(pFields, 0);

×

1823

  SField *pField1 = taosArrayGet(pFields, 1);

×

1825

  const char *oldTagName = pField0->name;

×

1826

  const char *newTagName = pField1->name;

×

1828

  int32_t tag = mndFindSuperTableTagIndex(pOld, oldTagName);

×

1829

  if (tag < 0) {

×

1830

    code = TSDB_CODE_MND_TAG_NOT_EXIST;

×

1831

    TAOS_RETURN(code);

×

1834

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

×

1835

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

×

1837

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

×

1838

    code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1839

    TAOS_RETURN(code);

×

1842

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

×

1843

    code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1844

    TAOS_RETURN(code);

×

1847

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1849

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

×

1850

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

×

1852

  pNew->tagVer++;

×

1853

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

×

1854

  TAOS_RETURN(code);

×

1857

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

×

1858

  int32_t code = 0;

×

1859

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

×

1860

  if (tag < 0) {

×

1861

    code = TSDB_CODE_MND_TAG_NOT_EXIST;

×

1862

    TAOS_RETURN(code);

×

1865

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

×

1866

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

×

1868

  uint32_t nLen = 0;

×

1869

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

×

1870

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

×

1873

  if (nLen > TSDB_MAX_TAGS_LEN) {

×

1874

    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;

×

1875

    TAOS_RETURN(code);

×

1878

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1880

  SSchema *pTag = pNew->pTags + tag;

×

1882

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

×

1883

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

×

1884

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

1885

    TAOS_RETURN(code);

×

1888

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

×

1889

    code = TSDB_CODE_MND_INVALID_ROW_BYTES;

×

1890

    TAOS_RETURN(code);

×

1893

  pTag->bytes = pField->bytes;

×

1894

  pNew->tagVer++;

×

1896

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

×

1897

  TAOS_RETURN(code);

×

1900

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

×

1904

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

×

1905

  TAOS_FIELD *p = taosArrayGet(pField, 0);

×

1907

  int32_t code = 0;

×

1908

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

×

1909

  if (idx == -1) {

×

1910

    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;

×

1911

    TAOS_RETURN(code);

×

1913

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

×

1914

  col_id_t colId = pTarget->colId;

×

1916

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1917

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

×

1918

  if (code != TSDB_CODE_SUCCESS) {

×

1919

    TAOS_RETURN(code);

×

1922

  int8_t updated = 0;

×

1923

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

×

1924

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

×

1925

    if (pCmpr->id == colId) {

×

1926

      uint32_t dst = 0;

×

1927

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

×

1929

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

×

1930

      break;

×

1934

  if (updated == 0) {

×

1935

    code = TSDB_CODE_MND_COLUMN_COMPRESS_ALREADY_EXIST;

×

1936

    TAOS_RETURN(code);

×

1937

  } else if (updated == -1) {

×

1938

    code = TSDB_CODE_TSC_COMPRESS_LEVEL_ERROR;

×

1939

    TAOS_RETURN(code);

×

1942

  pNew->colVer++;

×

1944

  TAOS_RETURN(code);

×

1946

static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, const SMAlterStbReq* pReq, int32_t ncols,

×

1948

  int32_t code = 0;

×

1949

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

×

1950

    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;

×

1951

    TAOS_RETURN(code);

×

1954

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

×

1955

    TAOS_RETURN(code);

×

1958

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

×

1959

    code = TSDB_CODE_PAR_INVALID_ROW_LENGTH;

×

1960

    TAOS_RETURN(code);

×

1963

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

×

1965

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

1967

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

×

1968

    code = TSDB_CODE_OUT_OF_RANGE;

×

1969

    TAOS_RETURN(code);

×

1972

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

×

1973

    if (withCompress) {

×

1974

      SFieldWithOptions *pField = taosArrayGet(pReq->pFields, i);

×

1975

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

×

1976

        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

1977

        TAOS_RETURN(code);

×

1980

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

×

1981

        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

1982

        TAOS_RETURN(code);

×

1985

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

×

1986

      pSchema->bytes = pField->bytes;

×

1987

      pSchema->type = pField->type;

×

1988

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

×

1989

      pSchema->colId = pNew->nextColId;

×

1990

      pNew->nextColId++;

×

1992

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

×

1993

      pCmpr->id = pSchema->colId;

×

1994

      pCmpr->alg = pField->compress;

×

1995

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

×

1997

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

×

1998

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

×

1999

        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;

×

2000

        TAOS_RETURN(code);

×

2003

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

×

2004

        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;

×

2005

        TAOS_RETURN(code);

×

2008

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

×

2009

      pSchema->bytes = pField->bytes;

×

2010

      pSchema->type = pField->type;

×

2011

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

×

2012

      pSchema->colId = pNew->nextColId;

×

2013

      pNew->nextColId++;

×

2015

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

×

2016

      pCmpr->id = pSchema->colId;

×

2017

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

×

2018

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

×

2023

  if (pReq->pTypeMods || pOld->pExtSchemas) {

×

2024

    if (!pNew->pExtSchemas) {

×

2026

      pNew->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));

×

2027

      if (!pNew->pExtSchemas) TAOS_RETURN(terrno);

×

2029

    if (pOld->pExtSchemas) {

×

2030

      memcpy(pNew->pExtSchemas, pOld->pExtSchemas, pOld->numOfColumns * sizeof(SExtSchema));

×

2032

    if (taosArrayGetSize(pReq->pTypeMods) > 0) {

×

2034

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

×

2035

        pNew->pColumns[pOld->numOfColumns + i].flags |= COL_HAS_TYPE_MOD;

×

2036

        pNew->pExtSchemas[pOld->numOfColumns + i].typeMod = *(STypeMod *)taosArrayGet(pReq->pTypeMods, i);

×

2041

  pNew->colVer++;

×

2042

  TAOS_RETURN(code);

×

2045

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

×

2046

  int32_t code = 0;

×

2047

  int32_t col = mndFindSuperTableColumnIndex(pOld, colName);

×

2048

  if (col < 0) {

×

2049

    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;

×

2050

    TAOS_RETURN(code);

×

2053

  if (col == 0) {

×

2054

    code = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;

×

2055

    TAOS_RETURN(code);

×

2058

  if (pOld->numOfColumns == 2) {

×

2059

    code = TSDB_CODE_PAR_INVALID_DROP_COL;

×

2060

    TAOS_RETURN(code);

×

2063

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

×

2064

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

×

2066

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

2068

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

×

2069

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

×

2070

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

×

2071

  if (pOld->pExtSchemas) {

×

2072

    memmove(pNew->pExtSchemas + col, pNew->pExtSchemas + col + 1, sizeof(SExtSchema) * sz);

×

2074

  pNew->numOfColumns--;

×

2076

  pNew->colVer++;

×

2077

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

×

2078

  TAOS_RETURN(code);

×

2081

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

×

2082

  int32_t code = 0;

×

2083

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

×

2084

  if (col < 0) {

×

2085

    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;

×

2086

    TAOS_RETURN(code);

×

2089

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

×

2091

  uint32_t nLen = 0;

×

2092

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

×

2093

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

×

2096

  if (nLen > TSDB_MAX_BYTES_PER_ROW) {

×

2097

    code = TSDB_CODE_MND_INVALID_ROW_BYTES;

×

2098

    TAOS_RETURN(code);

×

2101

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

×

2103

  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));

×

2105

  SSchema *pCol = pNew->pColumns + col;

×

2106

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

×

2107

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

×

2108

    code = TSDB_CODE_MND_INVALID_STB_OPTION;

×

2109

    TAOS_RETURN(code);

×

2112

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

×

2113

    code = TSDB_CODE_MND_INVALID_ROW_BYTES;

×

2114

    TAOS_RETURN(code);

×

2117

  pCol->bytes = pField->bytes;

×

2118

  pNew->colVer++;

×

2120

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

×

2121

  TAOS_RETURN(code);

×

2124

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

×

2125

  int32_t  code = 0;

×

2126

  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);

×

2127

  if (pRedoRaw == NULL) {

×

2128

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2129

    if (terrno != 0) code = terrno;

×

2130

    TAOS_RETURN(code);

×

2132

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

×

2133

    sdbFreeRaw(pRedoRaw);

×

2134

    TAOS_RETURN(code);

×

2136

  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY));

×

2138

  TAOS_RETURN(code);

×

2141

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

×

2142

  int32_t  code = 0;

×

2143

  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);

×

2144

  if (pCommitRaw == NULL) {

×

2145

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2146

    if (terrno != 0) code = terrno;

×

2147

    TAOS_RETURN(code);

×

2149

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

×

2150

    sdbFreeRaw(pCommitRaw);

×

2151

    TAOS_RETURN(code);

×

2153

  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

×

2155

  TAOS_RETURN(code);

×

2158

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

×

2160

  int32_t code = 0;

×

2161

  SSdb   *pSdb = pMnode->pSdb;

×

2162

  SVgObj *pVgroup = NULL;

×

2163

  void   *pIter = NULL;

×

2166

  while (1) {

×

2167

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

×

2168

    if (pIter == NULL) break;

×

2169

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

×

2170

      sdbRelease(pSdb, pVgroup);

×

2171

      continue;

×

2174

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

×

2175

    if (pReq == NULL) {

×

2176

      sdbCancelFetch(pSdb, pIter);

×

2177

      sdbRelease(pSdb, pVgroup);

×

2178

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2179

      if (terrno != 0) code = terrno;

×

2180

      TAOS_RETURN(code);

×

2182

    STransAction action = {0};

×

2183

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

2184

    action.pCont = pReq;

×

2185

    action.contLen = contLen;

×

2186

    action.msgType = TDMT_VND_ALTER_STB;

×

2187

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

×

2188

      taosMemoryFree(pReq);

×

2189

      sdbCancelFetch(pSdb, pIter);

×

2190

      sdbRelease(pSdb, pVgroup);

×

2191

      TAOS_RETURN(code);

×

2193

    sdbRelease(pSdb, pVgroup);

×

2196

  TAOS_RETURN(code);

×

2199

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

×

2201

  int32_t code = 0;

×

2202

  SSdb   *pSdb = pMnode->pSdb;

×

2203

  SVgObj *pVgroup = NULL;

×

2204

  void   *pIter = NULL;

×

2207

  while (1) {

×

2208

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

×

2209

    if (pIter == NULL) break;

×

2210

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

×

2211

      sdbRelease(pSdb, pVgroup);

×

2212

      continue;

×

2215

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

×

2216

    if (pReq == NULL) {

×

2217

      sdbCancelFetch(pSdb, pIter);

×

2218

      sdbRelease(pSdb, pVgroup);

×

2219

      code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2220

      if (terrno != 0) code = terrno;

×

2221

      TAOS_RETURN(code);

×

2223

    STransAction action = {0};

×

2224

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

2225

    action.pCont = pReq;

×

2226

    action.contLen = contLen;

×

2227

    action.msgType = TDMT_VND_CREATE_INDEX;

×

2228

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

×

2229

      taosMemoryFree(pReq);

×

2230

      sdbCancelFetch(pSdb, pIter);

×

2231

      sdbRelease(pSdb, pVgroup);

×

2232

      TAOS_RETURN(code);

×

2234

    sdbRelease(pSdb, pVgroup);

×

2237

  TAOS_RETURN(code);

×

2240

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

×

2241

  int32_t code = 0;

×

2242

  taosRLockLatch(&pStb->lock);

×

2244

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

×

2245

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

×

2246

  if (pRsp->pSchemas == NULL) {

×

2247

    taosRUnLockLatch(&pStb->lock);

×

2248

    code = terrno;

×

2249

    TAOS_RETURN(code);

×

2251

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

×

2252

  if (pRsp->pSchemaExt == NULL) {

×

2253

    taosRUnLockLatch(&pStb->lock);

×

2254

    code = terrno;

×

2255

    TAOS_RETURN(code);

×

2257

  pRsp->numOfColRefs = 0;

×

2258

  pRsp->pColRefs = NULL;

×

2259

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

×

2260

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

×

2261

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

×

2262

  pRsp->dbId = pDb->uid;

×

2263

  pRsp->numOfTags = pStb->numOfTags;

×

2264

  pRsp->numOfColumns = pStb->numOfColumns;

×

2265

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

×

2266

  pRsp->tableType = TSDB_SUPER_TABLE;

×

2267

  pRsp->sversion = pStb->colVer;

×

2268

  pRsp->tversion = pStb->tagVer;

×

2269

  pRsp->suid = pStb->uid;

×

2270

  pRsp->tuid = pStb->uid;

×

2271

  pRsp->virtualStb = pStb->virtualStb;

×

2273

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

×

2274

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

×

2275

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

×

2276

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

×

2277

    pSchema->type = pSrcSchema->type;

×

2278

    pSchema->flags = pSrcSchema->flags;

×

2279

    pSchema->colId = pSrcSchema->colId;

×

2280

    pSchema->bytes = pSrcSchema->bytes;

×

2283

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

×

2284

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

×

2285

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

×

2286

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

×

2287

    pSchema->type = pSrcSchema->type;

×

2288

    pSchema->flags = pSrcSchema->flags;

×

2289

    pSchema->colId = pSrcSchema->colId;

×

2290

    pSchema->bytes = pSrcSchema->bytes;

×

2292

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

×

2293

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

×

2294

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

×

2295

    pSchEx->colId = pCmpr->id;

×

2296

    pSchEx->compress = pCmpr->alg;

×

2297

    if (pStb->pExtSchemas) {

×

2298

      pSchEx->typeMod = pStb->pExtSchemas[i].typeMod;

×

2302

  taosRUnLockLatch(&pStb->lock);

×

2303

  TAOS_RETURN(code);

×

2306

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

×

2307

  int32_t code = 0;

×

2308

  taosRLockLatch(&pStb->lock);

×

2310

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

×

2311

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

×

2312

  if (pRsp->pSchemas == NULL) {

×

2313

    taosRUnLockLatch(&pStb->lock);

×

2314

    code = terrno;

×

2315

    TAOS_RETURN(code);

×

2318

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

×

2319

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

×

2320

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

×

2321

  pRsp->numOfTags = pStb->numOfTags;

×

2322

  pRsp->numOfColumns = pStb->numOfColumns;

×

2323

  pRsp->tableType = TSDB_SUPER_TABLE;

×

2324

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

×

2325

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

×

2326

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

×

2327

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

×

2328

  pRsp->ttl = pStb->ttl;

×

2329

  pRsp->keep = pStb->keep;

×

2330

  pRsp->commentLen = pStb->commentLen;

×

2331

  if (pStb->commentLen > 0) {

×

2332

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

×

2335

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

×

2336

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

×

2337

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

×

2338

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

×

2339

    pSchema->type = pSrcSchema->type;

×

2340

    pSchema->flags = pSrcSchema->flags;

×

2341

    pSchema->colId = pSrcSchema->colId;

×

2342

    pSchema->bytes = pSrcSchema->bytes;

×

2345

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

×

2346

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

×

2347

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

×

2348

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

×

2349

    pSchema->type = pSrcSchema->type;

×

2350

    pSchema->flags = pSrcSchema->flags;

×

2351

    pSchema->colId = pSrcSchema->colId;

×

2352

    pSchema->bytes = pSrcSchema->bytes;

×

2355

  if (pStb->numOfFuncs > 0) {

×

2356

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

×

2359

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

×

2360

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

×

2361

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

×

2363

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

×

2364

    pSchExt->colId = pCmpr->id;

×

2365

    pSchExt->compress = pCmpr->alg;

×

2366

    if (pStb->pExtSchemas) {

×

2367

      pSchExt->typeMod = pStb->pExtSchemas[i].typeMod;

×

2370

  pRsp->virtualStb = pStb->virtualStb;

×

2371

  pRsp->pColRefs = NULL;

×

2373

  taosRUnLockLatch(&pStb->lock);

×

2374

  TAOS_RETURN(code);

×

2377

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

×

2378

  int32_t code = 0;

×

2379

  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2380

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

×

2382

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

×

2383

  if (pDb == NULL) {

×

2384

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2385

    TAOS_RETURN(code);

×

2388

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

×

2389

    mndReleaseDb(pMnode, pDb);

×

2390

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2391

    TAOS_RETURN(code);

×

2394

  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);

×

2395

  if (pStb == NULL) {

×

2396

    mndReleaseDb(pMnode, pDb);

×

2397

    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

2398

    TAOS_RETURN(code);

×

2401

  taosRLockLatch(&pStb->lock);

×

2403

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

×

2404

    *schema = true;

×

2406

    *schema = false;

×

2409

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

×

2410

    *sma = true;

×

2412

    *sma = false;

×

2415

  taosRUnLockLatch(&pStb->lock);

×

2417

  mndReleaseDb(pMnode, pDb);

×

2418

  mndReleaseStb(pMnode, pStb);

×

2419

  return TSDB_CODE_SUCCESS;

×

2422

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

×

2423

  int32_t code = 0;

×

2424

  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2425

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

×

2427

  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);

×

2428

  if (pDb == NULL) {

×

2429

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2430

    TAOS_RETURN(code);

×

2433

  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);

×

2434

  if (pStb == NULL) {

×

2435

    mndReleaseDb(pMnode, pDb);

×

2436

    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

2437

    TAOS_RETURN(code);

×

2440

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

×

2441

  mndReleaseDb(pMnode, pDb);

×

2442

  mndReleaseStb(pMnode, pStb);

×

2443

  TAOS_RETURN(code);

×

2446

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

×

2447

  int32_t code = 0;

×

2448

  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};

×

2449

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

×

2451

  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);

×

2452

  if (pDb == NULL) {

×

2453

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

2454

    TAOS_RETURN(code);

×

2457

  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);

×

2458

  if (pStb == NULL) {

×

2459

    mndReleaseDb(pMnode, pDb);

×

2460

    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;

×

2461

    TAOS_RETURN(code);

×

2464

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

×

2466

  mndReleaseDb(pMnode, pDb);

×

2467

  mndReleaseStb(pMnode, pStb);

×

2468

  TAOS_RETURN(code);

×

2471

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

×

2472

  int32_t       code = 0;

×

2473

  SEncoder      ec = {0};

×

2474

  uint32_t      contLen = 0;

×

2475

  SMAlterStbRsp alterRsp = {0};

×

2476

  SName         name = {0};

×

2477

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

×

2479

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

×

2480

  if (NULL == alterRsp.pMeta) {

×

2481

    code = terrno;

×

2482

    TAOS_RETURN(code);

×

2485

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

×

2486

  if (code) {

×

2487

    tFreeSMAlterStbRsp(&alterRsp);

×

2488

    return code;

×

2491

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

×

2492

  if (code) {

×

2493

    tFreeSMAlterStbRsp(&alterRsp);

×

2494

    return code;

×

2497

  void *cont = taosMemoryMalloc(contLen);

×

2498

  if (NULL == cont) {

×

2499

    code = terrno;

×

2500

    tFreeSMAlterStbRsp(&alterRsp);

×

2501

    TAOS_RETURN(code);

×

2503

  tEncoderInit(&ec, cont, contLen);

×

2504

  code = tEncodeSMAlterStbRsp(&ec, &alterRsp);

×

2505

  tEncoderClear(&ec);

×

2507

  tFreeSMAlterStbRsp(&alterRsp);

×

2509

  if (code < 0) TAOS_RETURN(code);

×

2511

  *pCont = cont;

×

2512

  *pLen = contLen;

×

2514

  TAOS_RETURN(code);

×

2517

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

×

2518

  int32_t code = -1;

×

2519

  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);

×

2520

  if (NULL == pDb) {

×

2521

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2522

    if (terrno != 0) code = terrno;

×

2523

    TAOS_RETURN(code);

×

2526

  SStbObj *pObj = mndAcquireStb(pMnode, stbFName);

×

2527

  if (NULL == pObj) {

×

2528

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2529

    if (terrno != 0) code = terrno;

×

2530

    goto _OVER;

×

2533

  SEncoder       ec = {0};

×

2534

  uint32_t       contLen = 0;

×

2535

  SMCreateStbRsp stbRsp = {0};

×

2536

  SName          name = {0};

×

2537

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

×

2539

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

×

2540

  if (NULL == stbRsp.pMeta) {

×

2541

    code = terrno;

×

2542

    goto _OVER;

×

2545

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

×

2546

  if (code) {

×

2547

    tFreeSMCreateStbRsp(&stbRsp);

×

2548

    goto _OVER;

×

2551

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

×

2552

  if (code) {

×

2553

    tFreeSMCreateStbRsp(&stbRsp);

×

2554

    goto _OVER;

×

2557

  void *cont = taosMemoryMalloc(contLen);

×

2558

  if (NULL == cont) {

×

2559

    code = terrno;

×

2560

    tFreeSMCreateStbRsp(&stbRsp);

×

2561

    goto _OVER;

×

2563

  tEncoderInit(&ec, cont, contLen);

×

2564

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

×

2565

  tEncoderClear(&ec);

×

2567

  tFreeSMCreateStbRsp(&stbRsp);

×

2569

  *pCont = cont;

×

2570

  *pLen = contLen;

×

2572

  code = 0;

×

2574

_OVER:

×

2575

  if (pObj) {

×

2576

    mndReleaseStb(pMnode, pObj);

×

2579

  if (pDb) {

×

2580

    mndReleaseDb(pMnode, pDb);

×

2583

  TAOS_RETURN(code);

×

2586

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

×

2588

  int32_t code = -1;

×

2589

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

×

2590

  if (pTrans == NULL) {

×

2591

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2592

    if (terrno != 0) code = terrno;

×

2593

    goto _OVER;

×

2596

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

×

2597

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

×

2598

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

×

2600

  if (needRsp) {

×

2601

    void   *pCont = NULL;

×

2602

    int32_t contLen = 0;

×

2603

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

×

2604

    mndTransSetRpcRsp(pTrans, pCont, contLen);

×

2607

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

×

2608

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

×

2609

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

×

2610

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

×

2612

  code = 0;

×

2614

_OVER:

×

2615

  mndTransDrop(pTrans);

×

2616

  TAOS_RETURN(code);

×

2619

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

×

2621

  int32_t code = -1;

×

2622

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

×

2623

  if (pTrans == NULL) {

×

2624

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2625

    if (terrno != 0) code = terrno;

×

2626

    goto _OVER;

×

2629

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

×

2630

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

×

2632

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

×

2634

  if (needRsp) {

×

2635

    void   *pCont = NULL;

×

2636

    int32_t contLen = 0;

×

2637

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

×

2638

    mndTransSetRpcRsp(pTrans, pCont, contLen);

×

2641

  if (pAlter->alterType == TSDB_ALTER_TABLE_DROP_TAG) {

×

2642

    SIdxObj idxObj = {0};

×

2643

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

×

2644

    bool    exist = false;

×

2645

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

×

2646

      exist = true;

×

2648

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

×

2649

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

×

2651

    if (exist == true) {

×

2652

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

×

2653

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

×

2656

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

×

2657

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

×

2659

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

×

2660

    SIdxObj     idxObj = {0};

×

2661

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

×

2662

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

×

2663

    const char *oTagName = pField0->name;

×

2664

    const char *nTagName = pField1->name;

×

2665

    bool        exist = false;

×

2667

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

×

2668

      exist = true;

×

2671

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

×

2672

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

×

2674

    if (exist == true) {

×

2675

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

×

2676

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

×

2677

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

×

2678

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

×

2681

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

×

2682

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

×

2684

  code = 0;

×

2686

_OVER:

×

2687

  mndTransDrop(pTrans);

×

2688

  TAOS_RETURN(code);

×

2691

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

×

2692

  bool    needRsp = true;

×

2693

  int32_t code = -1;

×

2694

  SField *pField0 = NULL;

×

2696

  SStbObj stbObj = {0};

×

2697

  taosRLockLatch(&pOld->lock);

×

2698

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

×

2699

  taosRUnLockLatch(&pOld->lock);

×

2700

  stbObj.pColumns = NULL;

×

2701

  stbObj.pTags = NULL;

×

2702

  stbObj.pFuncs = NULL;

×

2703

  stbObj.pCmpr = NULL;

×

2704

  stbObj.pExtSchemas = NULL;

×

2705

  stbObj.updateTime = taosGetTimestampMs();

×

2706

  stbObj.lock = 0;

×

2707

  bool updateTagIndex = false;

×

2708

  switch (pAlter->alterType) {

×

2709

    case TSDB_ALTER_TABLE_ADD_TAG:

×

2710

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

×

2711

      break;

×

2712

    case TSDB_ALTER_TABLE_DROP_TAG:

×

2713

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

×

2714

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

×

2715

      updateTagIndex = true;

×

2716

      break;

×

2717

    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:

×

2718

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

×

2719

      updateTagIndex = true;

×

2720

      break;

×

2721

    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:

×

2722

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

×

2723

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

×

2724

      break;

×

2725

    case TSDB_ALTER_TABLE_ADD_COLUMN:

×

2726

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

×

2727

      break;

×

2728

    case TSDB_ALTER_TABLE_DROP_COLUMN:

×

2729

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

×

2730

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

×

2731

      break;

×

2732

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:

×

2733

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

×

2734

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

×

2735

      break;

×

2736

    case TSDB_ALTER_TABLE_UPDATE_OPTIONS:

×

2737

      needRsp = false;

×

2738

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

×

2739

      break;

×

2740

    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:

×

2741

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

×

2742

      break;

×

2743

    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION:

×

2744

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

×

2745

      break;

×

2746

    default:

×

2747

      needRsp = false;

×

2748

      terrno = TSDB_CODE_OPS_NOT_SUPPORT;

×

2749

      break;

×

2752

  if (code != 0) goto _OVER;

×

2753

  if (updateTagIndex == false) {

×

2754

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

×

2756

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

×

2759

_OVER:

×

2760

  taosMemoryFreeClear(stbObj.pTags);

×

2761

  taosMemoryFreeClear(stbObj.pColumns);

×

2762

  taosMemoryFreeClear(stbObj.pCmpr);

×

2763

  if (pAlter->commentLen > 0) {

×

2764

    taosMemoryFreeClear(stbObj.comment);

×

2766

  taosMemoryFreeClear(stbObj.pExtSchemas);

×

2767

  TAOS_RETURN(code);

×

2770

static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {

×

2771

  SMnode       *pMnode = pReq->info.node;

×

2772

  int32_t       code = -1;

×

2773

  SDbObj       *pDb = NULL;

×

2774

  SStbObj      *pStb = NULL;

×

2775

  SMAlterStbReq alterReq = {0};

×

2777

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

×

2778

    code = TSDB_CODE_INVALID_MSG;

×

2779

    goto _OVER;

×

2782

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

×

2783

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

×

2785

  pDb = mndAcquireDbByStb(pMnode, alterReq.name);

×

2786

  if (pDb == NULL) {

×

2787

    code = TSDB_CODE_MND_DB_NOT_EXIST;

×

2788

    goto _OVER;

×

2791

  pStb = mndAcquireStb(pMnode, alterReq.name);

×

2792

  if (pStb == NULL) {

×

2793

    code = TSDB_CODE_MND_STB_NOT_EXIST;

×

2794

    goto _OVER;

×

2797

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

×

2798

    goto _OVER;

×

2801

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

×

2802

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

2804

  SName   name = {0};

×

2805

  int32_t ret = 0;

×

2806

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

×

2807

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

×

2809

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

×

2811

_OVER:

×

2812

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

×

2813

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

×

2816

  mndReleaseStb(pMnode, pStb);

×

2817

  mndReleaseDb(pMnode, pDb);

×

2818

  tFreeSMAltertbReq(&alterReq);

×

2820

  TAOS_RETURN(code);

×

2823

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

×

2824

  int32_t  code = 0;

×

2825

  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);

×

2826

  if (pRedoRaw == NULL) {

×

2827

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2828

    if (terrno != 0) code = terrno;

×

2829

    TAOS_RETURN(code);

×

2831

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

×

2832

    sdbFreeRaw(pRedoRaw);

×

2833

    TAOS_RETURN(code);

×

2835

  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));

×

2837

  TAOS_RETURN(code);

×

2840

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

×

2841

  int32_t  code = 0;

×

2842

  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);

×

2843

  if (pCommitRaw == NULL) {

×

2844

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2845

    if (terrno != 0) code = terrno;

×

2846

    TAOS_RETURN(code);

×

2848

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

×

2849

    sdbFreeRaw(pCommitRaw);

×

2850

    TAOS_RETURN(code);

×

2852

  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

×

2854

  TAOS_RETURN(code);

×

2857

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

×

2858

  int32_t code = 0;

×

2859

  SSdb   *pSdb = pMnode->pSdb;

×

2860

  SVgObj *pVgroup = NULL;

×

2861

  void   *pIter = NULL;

×

2863

  while (1) {

×

2864

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

×

2865

    if (pIter == NULL) break;

×

2866

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

×

2867

      sdbRelease(pSdb, pVgroup);

×

2868

      continue;

×

2871

    int32_t contLen = 0;

×

2872

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

×

2873

    if (pReq == NULL) {

×

2874

      sdbCancelFetch(pSdb, pIter);

×

2875

      sdbRelease(pSdb, pVgroup);

×

2876

      code = TSDB_CODE_OUT_OF_MEMORY;

×

2877

      TAOS_RETURN(code);

×

2880

    STransAction action = {0};

×

2881

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

2882

    action.pCont = pReq;

×

2883

    action.contLen = contLen;

×

2884

    action.msgType = TDMT_VND_DROP_STB;

×

2885

    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;

×

2886

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

×

2887

      taosMemoryFree(pReq);

×

2888

      sdbCancelFetch(pSdb, pIter);

×

2889

      sdbRelease(pSdb, pVgroup);

×

2890

      TAOS_RETURN(code);

×

2892

    sdbRelease(pSdb, pVgroup);

×

2895

  TAOS_RETURN(code);

×

2898

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

×

2899

  int32_t code = -1;

×

2900

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

×

2901

  if (pTrans == NULL) {

×

2902

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

2903

    if (terrno != 0) code = terrno;

×

2904

    goto _OVER;

×

2907

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

×

2908

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

×

2909

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

×

2911

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

×

2912

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

×

2913

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

×

2914

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

×

2915

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

×

2916

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

×

2917

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

×

2918

  code = 0;

×

2920

_OVER:

×

2921

  mndTransDrop(pTrans);

×

2922

  TAOS_RETURN(code);

×

2925

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

×

2926

  int32_t code = 0;

×

2927

  SSdb   *pSdb = pMnode->pSdb;

×

2928

  void   *pIter = NULL;

×

2929

  while (1) {

×

2930

    SMqTopicObj *pTopic = NULL;

×

2931

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

×

2932

    if (pIter == NULL) break;

×

2934

    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {

×

2935

      if (pTopic->stbUid == suid) {

×

2936

        sdbRelease(pSdb, pTopic);

×

2937

        sdbCancelFetch(pSdb, pIter);

×

2938

        TAOS_RETURN(-1);

×

2942

    if (pTopic->ast == NULL) {

×

2943

      sdbRelease(pSdb, pTopic);

×

2944

      continue;

×

2947

    SNode *pAst = NULL;

×

2948

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

×

2949

      code = TSDB_CODE_MND_INVALID_TOPIC_OPTION;

×

2950

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

×

2951

      sdbRelease(pSdb, pTopic);

×

2952

      sdbCancelFetch(pSdb, pIter);

×

2953

      TAOS_RETURN(code);

×

2956

    SNodeList *pNodeList = NULL;

×

2957

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

×

2959

      sdbRelease(pSdb, pTopic);

×

2960

      sdbCancelFetch(pSdb, pIter);

×

2961

      TAOS_RETURN(code);

×

2963

    SNode *pNode = NULL;

×

2964

    FOREACH(pNode, pNodeList) {

×

2965

      SColumnNode *pCol = (SColumnNode *)pNode;

×

2967

      if (pCol->tableId == suid) {

×

2968

        sdbRelease(pSdb, pTopic);

×

2969

        nodesDestroyNode(pAst);

×

2970

        nodesDestroyList(pNodeList);

×

2971

        sdbCancelFetch(pSdb, pIter);

×

2972

        TAOS_RETURN(-1);

×

2974

        goto NEXT;

×

2977

  NEXT:

×

2978

    sdbRelease(pSdb, pTopic);

×

2979

    nodesDestroyNode(pAst);

×

2980

    nodesDestroyList(pNodeList);

×

2982

  TAOS_RETURN(code);

×

2985

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

×

2986

  int32_t code = 0;

×

2987

  SSdb   *pSdb = pMnode->pSdb;

×

2988

  void   *pIter = NULL;

×

2989

  while (1) {

×

2990

    SStreamObj *pStream = NULL;

×

2991

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

×

2992

    if (pIter == NULL) break;

×

2994

    if (pStream->targetStbUid == suid) {

×

2995

      sdbCancelFetch(pSdb, pIter);

×

2996

      sdbRelease(pSdb, pStream);

×

2997

      TAOS_RETURN(-1);

×

3000

    SNode *pAst = NULL;

×

3001

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

×

3002

      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;

×

3003

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

×

3004

      sdbCancelFetch(pSdb, pIter);

×

3005

      sdbRelease(pSdb, pStream);

×

3006

      TAOS_RETURN(code);

×

3009

    SNodeList *pNodeList = NULL;

×

3010

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

×

3012

      sdbCancelFetch(pSdb, pIter);

×

3013

      sdbRelease(pSdb, pStream);

×

3014

      TAOS_RETURN(code);

×

3016

    SNode *pNode = NULL;

×

3017

    FOREACH(pNode, pNodeList) {

×

3018

      SColumnNode *pCol = (SColumnNode *)pNode;

×

3020

      if (pCol->tableId == suid) {

×

3021

        sdbCancelFetch(pSdb, pIter);

×

3022

        sdbRelease(pSdb, pStream);

×

3023

        nodesDestroyNode(pAst);

×

3024

        nodesDestroyList(pNodeList);

×

3025

        TAOS_RETURN(-1);

×

3027

        goto NEXT;

×

3030

  NEXT:

×

3031

    sdbRelease(pSdb, pStream);

×

3032

    nodesDestroyNode(pAst);

×

3033

    nodesDestroyList(pNodeList);

×

3035

  TAOS_RETURN(code);

×

3038

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

×

3039

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

×

3040

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

×

3042

static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {

×

3043

  SMnode      *pMnode = pReq->info.node;

×

3044

  int32_t      code = -1;

×

3045

  SDbObj      *pDb = NULL;

×

3046

  SStbObj     *pStb = NULL;

×

3047

  SMDropStbReq dropReq = {0};

×

3049

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

×

3051

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

×

3053

  pStb = mndAcquireStb(pMnode, dropReq.name);

×

3054

  if (pStb == NULL) {

×

3055

    if (dropReq.igNotExists) {

×

3056

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

×

3057

      code = 0;

×

3058

      goto _OVER;

×

3060

      code = TSDB_CODE_MND_STB_NOT_EXIST;

×

3061

      goto _OVER;

×

3065

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

×

3066

    code = 0;

×

3067

    goto _OVER;

×

3070

  pDb = mndAcquireDbByStb(pMnode, dropReq.name);

×

3071

  if (pDb == NULL) {

×

3072

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

3073

    goto _OVER;

×

3076

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

×

3077

    goto _OVER;

×

3080

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

×

3081

    code = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;

×

3082

    goto _OVER;

×

3085

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

×

3086

    code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;

×

3087

    goto _OVER;

×

3090

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

×

3091

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

3093

  SName   name = {0};

×

3094

  int32_t ret = 0;

×

3095

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

×

3096

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

×

3098

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

×

3100

_OVER:

×

3101

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

×

3102

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

×

3105

  mndReleaseDb(pMnode, pDb);

×

3106

  mndReleaseStb(pMnode, pStb);

×

3107

  tFreeSMDropStbReq(&dropReq);

×

3108

  TAOS_RETURN(code);

×

3111

static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {

×

3112

  SMnode       *pMnode = pReq->info.node;

×

3113

  int32_t       code = -1;

×

3114

  STableInfoReq infoReq = {0};

×

3115

  STableMetaRsp metaRsp = {0};

×

3116

  SUserObj     *pUser = NULL;

×

3118

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

×

3119

  if (pUser == NULL) return 0;

×

3120

  bool sysinfo = pUser->sysInfo;

×

3122

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

×

3124

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

×

3125

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

×

3126

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

×

3127

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

×

3128

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

×

3129

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

×

3131

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

×

3132

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

×

3135

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

×

3136

  if (rspLen < 0) {

×

3137

    code = TSDB_CODE_INVALID_MSG;

×

3138

    goto _OVER;

×

3141

  void *pRsp = rpcMallocCont(rspLen);

×

3142

  if (pRsp == NULL) {

×

3143

    code = terrno;

×

3144

    goto _OVER;

×

3147

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

×

3148

    code = rspLen;

×

3149

    goto _OVER;

×

3151

  pReq->info.rsp = pRsp;

×

3152

  pReq->info.rspLen = rspLen;

×

3153

  code = 0;

×

3155

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

×

3157

_OVER:

×

3158

  if (code != 0) {

×

3159

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

×

3162

  mndReleaseUser(pMnode, pUser);

×

3163

  tFreeSTableMetaRsp(&metaRsp);

×

3165

  return code;

×

3168

static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {

×

3169

  SMnode      *pMnode = pReq->info.node;

×

3170

  int32_t      code = -1;

×

3171

  STableCfgReq cfgReq = {0};

×

3172

  STableCfgRsp cfgRsp = {0};

×

3174

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

×

3176

  char dbName[TSDB_DB_NAME_LEN] = {0};

×

3177

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

×

3178

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

×

3179

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

×

3180

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

×

3181

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

×

3182

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

×

3183

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

×

3185

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

×

3186

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

×

3189

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

×

3190

  if (rspLen < 0) {

×

3191

    code = TSDB_CODE_INVALID_MSG;

×

3192

    goto _OVER;

×

3195

  void *pRsp = rpcMallocCont(rspLen);

×

3196

  if (pRsp == NULL) {

×

3197

    code = terrno;

×

3198

    goto _OVER;

×

3201

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

×

3202

    code = rspLen;

×

3203

    goto _OVER;

×

3205

  pReq->info.rsp = pRsp;

×

3206

  pReq->info.rspLen = rspLen;

×

3207

  code = 0;

×

3209

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

×

3211

_OVER:

×

3212

  if (code != 0) {

×

3213

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

×

3216

  tFreeSTableCfgRsp(&cfgRsp);

×

3217

  TAOS_RETURN(code);

×

3220

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

×

3222

  int32_t   code = 0;

×

3223

  SSTbHbRsp hbRsp = {0};

×

3224

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

×

3225

  if (hbRsp.pMetaRsp == NULL) {

×

3226

    code = terrno;

×

3227

    TAOS_RETURN(code);

×

3230

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

×

3231

  if (NULL == hbRsp.pIndexRsp) {

×

3232

    taosArrayDestroy(hbRsp.pMetaRsp);

×

3233

    code = terrno;

×

3234

    TAOS_RETURN(code);

×

3237

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

×

3238

    SSTableVersion *pStbVersion = &pStbVersions[i];

×

3239

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

×

3240

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

×

3241

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

×

3242

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

×

3244

    bool    schema = false;

×

3245

    bool    sma = false;

×

3246

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

×

3247

    if (TSDB_CODE_SUCCESS != code) {

×

3248

      STableMetaRsp metaRsp = {0};

×

3249

      metaRsp.numOfColumns = -1;

×

3250

      metaRsp.suid = pStbVersion->suid;

×

3251

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

×

3252

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

×

3253

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

×

3254

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

×

3255

        code = terrno;

×

3256

        return code;

×

3258

      continue;

×

3261

    if (schema) {

×

3262

      STableMetaRsp metaRsp = {0};

×

3263

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

×

3264

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

×

3265

        metaRsp.numOfColumns = -1;

×

3266

        metaRsp.suid = pStbVersion->suid;

×

3267

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

×

3268

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

×

3269

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

×

3270

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

×

3271

          code = terrno;

×

3272

          return code;

×

3274

        continue;

×

3277

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

×

3278

        code = terrno;

×

3279

        return code;

×

3283

    if (sma) {

×

3284

      bool           exist = false;

×

3286

      STableIndexRsp indexRsp = {0};

×

3287

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

×

3288

      if (NULL == indexRsp.pIndex) {

×

3289

        code = terrno;

×

3290

        TAOS_RETURN(code);

×

3293

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

×

3294

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

×

3295

      if (code || !exist) {

×

3296

        indexRsp.suid = pStbVersion->suid;

×

3297

        indexRsp.version = -1;

×

3298

        indexRsp.pIndex = NULL;

×

3301

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

×

3302

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

×

3304

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

×

3305

        code = terrno;

×

3306

        return code;

×

3311

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

×

3312

  if (rspLen < 0) {

×

3313

    tFreeSSTbHbRsp(&hbRsp);

×

3314

    code = TSDB_CODE_INVALID_MSG;

×

3315

    TAOS_RETURN(code);

×

3318

  void *pRsp = taosMemoryMalloc(rspLen);

×

3319

  if (pRsp == NULL) {

×

3320

    tFreeSSTbHbRsp(&hbRsp);

×

3321

    code = terrno;

×

3322

    TAOS_RETURN(code);

×

3325

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

×

3326

  tFreeSSTbHbRsp(&hbRsp);

×

3327

  if (rspLen < 0) return rspLen;

×

3328

  *ppRsp = pRsp;

×

3329

  *pRspLen = rspLen;

×

3330

  TAOS_RETURN(code);

×

3333

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

×

3334

  int32_t code = 0;

×

3335

  SSdb   *pSdb = pMnode->pSdb;

×

3336

  SDbObj *pDb = mndAcquireDb(pMnode, dbName);

×

3337

  if (pDb == NULL) {

×

3338

    code = TSDB_CODE_MND_DB_NOT_SELECTED;

×

3339

    TAOS_RETURN(code);

×

3342

  int32_t numOfStbs = 0;

×

3343

  void   *pIter = NULL;

×

3344

  while (1) {

×

3345

    SStbObj *pStb = NULL;

×

3346

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

×

3347

    if (pIter == NULL) break;

×

3349

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

×

3350

      numOfStbs++;

×

3353

    sdbRelease(pSdb, pStb);

×

3356

  *pNumOfStbs = numOfStbs;

×

3357

  mndReleaseDb(pMnode, pDb);

×

3358

  TAOS_RETURN(code);

×

3361

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

×

3362

  SName name = {0};

×

3363

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

×

3365

  TAOS_CHECK_RETURN(tNameGetFullDbName(&name, dst));

×

3367

  return 0;

×

3370

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

×

3371

  SName name = {0};

×

3372

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

×

3374

  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));

×

3376

  return 0;

×

3379

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

×

3380

  SName name = {0};

×

3381

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

×

3383

  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));

×

3385

  return 0;

×

3388

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

×

3389

  int32_t pos = -1;

×

3390

  int32_t num = 0;

×

3391

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

×

3392

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

×

3393

    if (num == 2) break;

×

3396

  if (num == 2) {

×

3397

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

×

3399

×

3401

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

×

3402

  SMnode  *pMnode = pReq->info.node;

×

3403

  SSdb    *pSdb = pMnode->pSdb;

×

3404

  int32_t  numOfRows = 0;

×

3405

  SStbObj *pStb = NULL;

×

3406

  int32_t  cols = 0;

×

3407

  int32_t  lino = 0;

×

3408

  int32_t  code = 0;

×

3410

  SDbObj *pDb = NULL;

×

3411

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

×

3412

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

×

3413

    if (pDb == NULL) return terrno;

×

3416

  while (numOfRows < rows) {

×

3417

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

×

3418

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

×

3420

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

×

3421

      sdbRelease(pSdb, pStb);

×

3422

      continue;

×

3425

    if (isTsmaResSTb(pStb->name)) {

×

3426

      sdbRelease(pSdb, pStb);

×

3427

      continue;

×

3430

    cols = 0;

×

3432

    SName name = {0};

×

3434

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

×

3435

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

×

3436

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

×

3437

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

×

3438

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

×

3440

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

×

3441

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

×

3442

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

×

3443

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

×

3444

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

×

3445

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

×

3447

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

×

3448

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

×

3451

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

×

3452

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

×

3455

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

×

3456

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

×

3458

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

×

3459

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

×

3462

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

×

3463

    if (pStb->commentLen > 0) {

×

3464

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

×

3465

      STR_TO_VARSTR(comment, pStb->comment);

×

3466

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

×

3467

    } else if (pStb->commentLen == 0) {

×

3468

      char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};

×

3469

      STR_TO_VARSTR(comment, "");

×

3470

      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);

×

3472

      colDataSetNULL(pColInfo, numOfRows);

×

3475

    char watermark[64 + VARSTR_HEADER_SIZE] = {0};

×

3476

    (void)tsnprintf(varDataVal(watermark), sizeof(watermark) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",

×

3477

              pStb->watermark[0], pStb->watermark[1]);

×

3478

    varDataSetLen(watermark, strlen(varDataVal(watermark)));

×

3480

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3481

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false), pStb, &lino, _ERROR);

×

3483

    char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};

×

3484

    (void)tsnprintf(varDataVal(maxDelay), sizeof(maxDelay) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",

×

3485

              pStb->maxdelay[0], pStb->maxdelay[1]);

×

3486

    varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));

×

3488

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3489

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false), pStb, &lino, _ERROR);

×

3491

    char    rollup[160 + VARSTR_HEADER_SIZE] = {0};

×

3492

    int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);

×

3493

    char   *sep = ", ";

×

3494

    int32_t sepLen = strlen(sep);

×

3495

    int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;

×

3496

    for (int32_t i = 0; i < rollupNum; ++i) {

×

3497

      char *funcName = taosArrayGet(pStb->pFuncs, i);

×

3498

      if (i) {

×

3499

        (void)strncat(varDataVal(rollup), sep, rollupLen);

×

3500

        rollupLen -= sepLen;

×

3502

      (void)strncat(varDataVal(rollup), funcName, rollupLen);

×

3503

      rollupLen -= strlen(funcName);

×

3505

    varDataSetLen(rollup, strlen(varDataVal(rollup)));

×

3507

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3508

    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);

×

3510

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3511

    if (pColInfo) {

×

3512

      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->uid), false), pStb, &lino, _ERROR);

×

3515

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3516

    if (pColInfo) {

×

3517

      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->virtualStb), false), pStb, &lino, _ERROR);

×

3519

    numOfRows++;

×

3520

    sdbRelease(pSdb, pStb);

×

3523

  if (pDb != NULL) {

×

3524

    mndReleaseDb(pMnode, pDb);

×

3527

  goto _OVER;

×

3529

_ERROR:

×

3530

  mError("show:0x%" PRIx64 ", failed to retrieve data at %s:%d since %s", pShow->id, __FUNCTION__, lino,

×

3533

_OVER:

×

3534

  pShow->numOfRows += numOfRows;

×

3535

  return numOfRows;

×

3538

static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *pSysDbTableMeta, size_t size,

×

3540

  char    tName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3541

  char    dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3542

  char    typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3543

  int32_t numOfRows = p->info.rows;

×

3544

  int32_t lino = 0;

×

3545

  int32_t code = 0;

×

3547

  STR_TO_VARSTR(dName, dbName);

×

3548

  STR_TO_VARSTR(typeName, "SYSTEM_TABLE");

×

3550

  for (int32_t i = 0; i < size; ++i) {

×

3551

    const SSysTableMeta *pm = &pSysDbTableMeta[i];

×

3555

    if (tbName[0] && strncmp(tbName, pm->name, TSDB_TABLE_NAME_LEN) != 0) {

×

3556

      continue;

×

3559

    STR_TO_VARSTR(tName, pm->name);

×

3561

    for (int32_t j = 0; j < pm->colNum; j++) {

×

3563

      SColumnInfoData *pColInfoData = taosArrayGet(p->pDataBlock, 0);

×

3564

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, tName, false), &lino, _OVER);

×

3567

      pColInfoData = taosArrayGet(p->pDataBlock, 1);

×

3568

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, dName, false), &lino, _OVER);

×

3570

      pColInfoData = taosArrayGet(p->pDataBlock, 2);

×

3571

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, typeName, false), &lino, _OVER);

×

3574

      char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3575

      STR_TO_VARSTR(colName, pm->schema[j].name);

×

3576

      pColInfoData = taosArrayGet(p->pDataBlock, 3);

×

3577

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, colName, false), &lino, _OVER);

×

3580

      int8_t colType = pm->schema[j].type;

×

3581

      pColInfoData = taosArrayGet(p->pDataBlock, 4);

×

3583

      int  colTypeLen =

×

3584

          tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);

×

3585

      if (colType == TSDB_DATA_TYPE_VARCHAR) {

×

3586

        colTypeLen +=

×

3587

            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",

×

3588

                      (int32_t)(pm->schema[j].bytes - VARSTR_HEADER_SIZE));

×

3589

      } else if (colType == TSDB_DATA_TYPE_NCHAR) {

×

3590

        colTypeLen +=

×

3591

            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",

×

3592

                      (int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));

×

3594

      varDataSetLen(colTypeStr, colTypeLen);

×

3595

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false), &lino, _OVER);

×

3597

      pColInfoData = taosArrayGet(p->pDataBlock, 5);

×

3598

      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false), &lino, _OVER);

×

3599

      for (int32_t k = 6; k <= 9; ++k) {

×

3600

        pColInfoData = taosArrayGet(p->pDataBlock, k);

×

3601

        colDataSetNULL(pColInfoData, numOfRows);

×

3604

      numOfRows += 1;

×

3607

  return numOfRows;

×

3608

_OVER:

×

3609

  mError("failed at %s:%d since %s", __FUNCTION__, lino, tstrerror(code));

×

3610

  return numOfRows;

×

3617

static int32_t buildSysDbColsInfo(SSDataBlock *p, int8_t buildWhichDBs, char *tb) {

×

3618

  size_t               size = 0;

×

3619

  const SSysTableMeta *pSysDbTableMeta = NULL;

×

3621

  if (buildWhichDBs & BUILD_COL_FOR_INFO_DB) {

×

3622

    getInfosDbMeta(&pSysDbTableMeta, &size);

×

3623

    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB, tb);

×

3626

  if (buildWhichDBs & BUILD_COL_FOR_PERF_DB) {

×

3627

    getPerfDbMeta(&pSysDbTableMeta, &size);

×

3628

    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB, tb);

×

3631

  return p->info.rows;

×

3634

static int8_t determineBuildColForWhichDBs(const char *db) {

×

3636

  if (!db[0])

×

3637

    buildWhichDBs = BUILD_COL_FOR_ALL_DB;

×

3639

    char *p = strchr(db, '.');

×

3640

    if (p && strcmp(p + 1, TSDB_INFORMATION_SCHEMA_DB) == 0) {

×

3641

      buildWhichDBs = BUILD_COL_FOR_INFO_DB;

×

3642

    } else if (p && strcmp(p + 1, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {

×

3643

      buildWhichDBs = BUILD_COL_FOR_PERF_DB;

×

3645

      buildWhichDBs = BUILD_COL_FOR_USER_DB;

×

3648

  return buildWhichDBs;

×

3651

static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {

×

3653

  SMnode  *pMnode = pReq->info.node;

×

3654

  SSdb    *pSdb = pMnode->pSdb;

×

3655

  SStbObj *pStb = NULL;

×

3656

  int32_t  numOfRows = 0;

×

3657

  int32_t  lino = 0;

×

3658

  int32_t  code = 0;

×

3660

  buildWhichDBs = determineBuildColForWhichDBs(pShow->db);

×

3662

  if (!pShow->sysDbRsp) {

×

3663

    numOfRows = buildSysDbColsInfo(pBlock, buildWhichDBs, pShow->filterTb);

×

3664

    mDebug("mndRetrieveStbCol get system table cols, rows:%d, db:%s", numOfRows, pShow->db);

×

3665

    pShow->sysDbRsp = true;

×

3668

  if (buildWhichDBs & BUILD_COL_FOR_USER_DB) {

×

3669

    SDbObj *pDb = NULL;

×

3670

    if (strlen(pShow->db) > 0) {

×

3671

      pDb = mndAcquireDb(pMnode, pShow->db);

×

3672

      if (pDb == NULL && TSDB_CODE_MND_DB_NOT_EXIST != terrno && pBlock->info.rows == 0) return terrno;

×

3675

    char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3676

    STR_TO_VARSTR(typeName, "SUPER_TABLE");

×

3677

    bool fetch = pShow->restore ? false : true;

×

3678

    pShow->restore = false;

×

3679

    while (numOfRows < rows) {

×

3680

      if (fetch) {

×

3681

        pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);

×

3682

        if (pShow->pIter == NULL) break;

×

3684

        fetch = true;

×

3685

        void *pKey = taosHashGetKey(pShow->pIter, NULL);

×

3686

        pStb = sdbAcquire(pSdb, SDB_STB, pKey);

×

3687

        if (!pStb) continue;

×

3690

      if (pDb != NULL && pStb->dbUid != pDb->uid) {

×

3691

        sdbRelease(pSdb, pStb);

×

3692

        continue;

×

3695

      SName name = {0};

×

3696

      char  stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3697

      mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);

×

3698

      if (pShow->filterTb[0] && strncmp(pShow->filterTb, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN) != 0) {

×

3699

        sdbRelease(pSdb, pStb);

×

3700

        continue;

×

3703

      if ((numOfRows + pStb->numOfColumns) > rows) {

×

3704

        pShow->restore = true;

×

3705

        if (numOfRows == 0) {

×

3706

          mError("mndRetrieveStbCol failed to get stable cols since buf:%d less than result:%d, stable name:%s, db:%s",

×

3709

        sdbRelease(pSdb, pStb);

×

3710

        break;

×

3713

      varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));

×

3715

      mDebug("mndRetrieveStbCol get stable cols, stable name:%s, db:%s", pStb->name, pStb->db);

×

3717

      char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3718

      RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _OVER);

×

3719

      RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _OVER);

×

3720

      varDataSetLen(db, strlen(varDataVal(db)));

×

3722

      for (int i = 0; i < pStb->numOfColumns; i++) {

×

3723

        int32_t          cols = 0;

×

3724

        SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3725

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _OVER);

×

3727

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3728

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _OVER);

×

3730

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3731

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeName, false), pStb, &lino, _OVER);

×

3734

        char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

3735

        STR_TO_VARSTR(colName, pStb->pColumns[i].name);

×

3736

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3737

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, colName, false), pStb, &lino, _OVER);

×

3740

        int8_t colType = pStb->pColumns[i].type;

×

3741

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3743

        int  colTypeLen =

×

3744

            tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);

×

3745

        if (colType == TSDB_DATA_TYPE_VARCHAR) {

×

3746

          colTypeLen +=

×

3747

              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,

×

3748

                        "(%d)", (int32_t)(pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE));

×

3749

        } else if (colType == TSDB_DATA_TYPE_NCHAR) {

×

3750

          colTypeLen +=

×

3751

              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,

×

3752

                        "(%d)", (int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));

×

3753

        } else if (IS_DECIMAL_TYPE(colType)) {

×

3754

          STypeMod typeMod = pStb->pExtSchemas[i].typeMod;

×

3755

          uint8_t prec = 0, scale = 0;

×

3756

          decimalFromTypeMod(typeMod, &prec, &scale);

×

3757

          colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d,%d)", prec, scale);

×

3759

        varDataSetLen(colTypeStr, colTypeLen);

×

3760

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false), pStb, &lino, _OVER);

×

3762

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3763

        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false), pStb,

×

3765

        while (cols < pShow->numOfColumns) {

×

3766

          pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

3767

          colDataSetNULL(pColInfo, numOfRows);

×

3769

        numOfRows++;

×

3772

      sdbRelease(pSdb, pStb);

×

3775

    if (pDb != NULL) {

×

3776

      mndReleaseDb(pMnode, pDb);

×

3780

  mDebug("mndRetrieveStbCol success, rows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows);

×

3781

  goto _OVER;

×

3787

_OVER:

×

3788

  pShow->numOfRows += numOfRows;

×

3789

  return numOfRows;

×

3792

static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {

×

3793

  SSdb *pSdb = pMnode->pSdb;

×

3794

  sdbCancelFetchByType(pSdb, pIter, SDB_STB);

×

3795

×

3797

const char *mndGetStbStr(const char *src) {

×

3798

  char *posDb = strstr(src, TS_PATH_DELIMITER);

×

3799

  if (posDb != NULL) ++posDb;

×

3800

  if (posDb == NULL) return src;

×

3802

  char *posStb = strstr(posDb, TS_PATH_DELIMITER);

×

3803

  if (posStb != NULL) ++posStb;

×

3804

  if (posStb == NULL) return posDb;

×

3805

  return posStb;

×

3808

static int32_t mndCheckIndexReq(SCreateTagIndexReq *pReq) {

×

3810

  return TSDB_CODE_SUCCESS;

×

3937

static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq) {

×

3938

  int32_t code = mndProcessDropStbReq(pReq);

×

3939

  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

×

3940

    pReq->info.rsp = rpcMallocCont(1);

×

3941

    pReq->info.rspLen = 1;

×

3942

    pReq->info.noResp = false;

×

3943

    pReq->code = code;

×

3945

  return code;

×

3978

static void mndDestroyDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx *p) {

×

3979

  if (!p) return;

×

3981

  if (p->pVgMap) {

×

3982

    void *pIter = taosHashIterate(p->pVgMap, NULL);

×

3983

    while (pIter) {

×

3984

      SVDropTbVgReqs *pReqs = pIter;

×

3985

      taosArrayDestroyEx(pReqs->pBatchReqs, destroySVDropTbBatchReqs);

×

3986

      pIter = taosHashIterate(p->pVgMap, pIter);

×

3988

    taosHashCleanup(p->pVgMap);

×

3990

  taosMemoryFree(p);

×

3993

static int32_t mndInitDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx **ppCtx) {

×

3994

  int32_t                 code = 0;

×

3995

  SMndDropTbsWithTsmaCtx *pCtx = taosMemoryCalloc(1, sizeof(SMndDropTbsWithTsmaCtx));

×

3996

  if (!pCtx) return terrno;

×

3998

  pCtx->pVgMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);

×

3999

  if (!pCtx->pVgMap) {

×

4000

    code = terrno;

×

4001

    goto _end;

×

4004

  *ppCtx = pCtx;

×

4005

_end:

×

4006

  if (code) mndDestroyDropTbsWithTsmaCtx(pCtx);

×

4007

  return code;

×

4010

static void *mndBuildVDropTbsReq(SMnode *pMnode, const SVgroupInfo *pVgInfo, const SVDropTbBatchReq *pReq,

×

4012

  int32_t   contLen = 0;

×

4013

  int32_t   ret = 0;

×

4014

  SMsgHead *pHead = NULL;

×

4015

  SEncoder  encoder = {0};

×

4017

  tEncodeSize(tEncodeSVDropTbBatchReq, pReq, contLen, ret);

×

4018

  if (ret < 0) return NULL;

×

4020

  contLen += sizeof(SMsgHead);

×

4021

  pHead = taosMemoryMalloc(contLen);

×

4022

  if (pHead == NULL) {

×

4023

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

4024

    return NULL;

×

4027

  pHead->contLen = htonl(contLen);

×

4028

  pHead->vgId = htonl(pVgInfo->vgId);

×

4030

  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));

×

4032

  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));

×

4033

  int32_t code = tEncodeSVDropTbBatchReq(&encoder, pReq);

×

4034

  tEncoderClear(&encoder);

×

4035

  if (code != 0) return NULL;

×

4037

  *len = contLen;

×

4038

  return pHead;

×

4041

static int32_t mndSetDropTbsRedoActions(SMnode *pMnode, STrans *pTrans, const SVDropTbVgReqs *pVgReqs, void *pCont,

×

4043

  STransAction action = {0};

×

4044

  action.epSet = pVgReqs->info.epSet;

×

4045

  action.pCont = pCont;

×

4046

  action.contLen = contLen;

×

4047

  action.msgType = msgType;

×

4048

  action.acceptableCode = TSDB_CODE_TDB_TABLE_NOT_EXIST;

×

4049

  return mndTransAppendRedoAction(pTrans, &action);

×

4052

static int32_t mndBuildDropTbRedoActions(SMnode *pMnode, STrans *pTrans, SHashObj *pVgMap, tmsg_t msgType) {

×

4053

  int32_t code = 0;

×

4054

  void   *pIter = taosHashIterate(pVgMap, NULL);

×

4055

  while (pIter) {

×

4056

    const SVDropTbVgReqs *pVgReqs = pIter;

×

4057

    int32_t               len = 0;

×

4058

    for (int32_t i = 0; i < taosArrayGetSize(pVgReqs->pBatchReqs) && code == TSDB_CODE_SUCCESS; ++i) {

×

4059

      SVDropTbBatchReq *pBatchReq = taosArrayGet(pVgReqs->pBatchReqs, i);

×

4060

      void             *p = mndBuildVDropTbsReq(pMnode, &pVgReqs->info, pBatchReq, &len);

×

4061

      if (!p) {

×

4062

        code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

4063

        if (terrno != 0) code = terrno;

×

4064

        break;

×

4066

      if ((code = mndSetDropTbsRedoActions(pMnode, pTrans, pVgReqs, p, len, msgType)) != 0) {

×

4067

        break;

×

4070

    if (TSDB_CODE_SUCCESS != code) {

×

4071

      taosHashCancelIterate(pVgMap, pIter);

×

4072

      break;

×

4074

    pIter = taosHashIterate(pVgMap, pIter);

×

4076

  return code;

×

4079

static int32_t mndCreateDropTbsTxnPrepare(SRpcMsg *pRsp, SMndDropTbsWithTsmaCtx *pCtx) {

×

4080

  int32_t code = 0;

×

4081

  SMnode *pMnode = pRsp->info.node;

×

4082

  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pRsp, "drop-tbs");

×

4083

  mndTransSetChangeless(pTrans);

×

4084

  mndTransSetSerial(pTrans);

×

4085

  if (pTrans == NULL) {

×

4086

    code = TSDB_CODE_MND_RETURN_VALUE_NULL;

×

4087

    if (terrno != 0) code = terrno;

×

4088

    goto _OVER;

×

4091

  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);

×

4093

  if ((code = mndBuildDropTbRedoActions(pMnode, pTrans, pCtx->pVgMap, TDMT_VND_DROP_TABLE)) != 0) goto _OVER;

×

4094

  if ((code = mndTransPrepare(pMnode, pTrans)) != 0) goto _OVER;

×

4096

_OVER:

×

4097

  mndTransDrop(pTrans);

×

4098

  TAOS_RETURN(code);

×

4101

static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq) {

×

4102

  int32_t      code = -1;

×

4103

  SMnode      *pMnode = pReq->info.node;

×

4104

  SDbObj      *pDb = NULL;

×

4105

  SStbObj     *pStb = NULL;

×

4106

  SMDropTbsReq dropReq = {0};

×

4107

  bool         locked = false;

×

4108

  if (tDeserializeSMDropTbsReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {

×

4109

    code = TSDB_CODE_INVALID_MSG;

×

4110

    goto _OVER;

×

4113

  SMndDropTbsWithTsmaCtx *pCtx = NULL;

×

4114

  code = mndInitDropTbsWithTsmaCtx(&pCtx);

×

4115

  if (code) goto _OVER;

×

4116

  for (int32_t i = 0; i < dropReq.pVgReqs->size; ++i) {

×

4117

    SMDropTbReqsOnSingleVg *pReq = taosArrayGet(dropReq.pVgReqs, i);

×

4118

    code = mndDropTbForSingleVg(pMnode, pCtx, pReq->pTbs, pReq->vgInfo.vgId);

×

4119

    if (code) goto _OVER;

×

4121

  code = mndCreateDropTbsTxnPrepare(pReq, pCtx);

×

4122

  if (code == 0) {

×

4123

    code = TSDB_CODE_ACTION_IN_PROGRESS;

×

4125

_OVER:

×

4126

  tFreeSMDropTbsReq(&dropReq);

×

4127

  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);

×

4128

  TAOS_RETURN(code);

×

4131

static int32_t createDropTbBatchReq(const SVDropTbReq *pReq, SVDropTbBatchReq *pBatchReq) {

×

4132

  pBatchReq->nReqs = 1;

×

4133

  pBatchReq->pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));

×

4134

  if (!pBatchReq->pArray) return terrno;

×

4135

  if (taosArrayPush(pBatchReq->pArray, pReq) == NULL) {

×

4136

    taosArrayDestroy(pBatchReq->pArray);

×

4137

    pBatchReq->pArray = NULL;

×

4138

    return terrno;

×

4140

  return TSDB_CODE_SUCCESS;

×

4143

static void destroySVDropTbBatchReqs(void *p) {

×

4144

  SVDropTbBatchReq *pReq = p;

×

4145

  taosArrayDestroy(pReq->pArray);

×

4146

  pReq->pArray = NULL;

×

4147

×

4149

static int32_t mndDropTbAdd(SMnode *pMnode, SHashObj *pVgHashMap, const SVgroupInfo *pVgInfo, char *name, tb_uid_t suid,

×

4151

  SVDropTbReq req = {.name = name, .suid = suid, .igNotExists = ignoreNotExists, .uid = 0};

×

4153

  SVDropTbVgReqs *pVgReqs = taosHashGet(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));

×

4154

  SVDropTbVgReqs  vgReqs = {0};

×

4155

  if (pVgReqs == NULL) {

×

4156

    vgReqs.info = *pVgInfo;

×

4157

    vgReqs.pBatchReqs = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbBatchReq));

×

4158

    if (!vgReqs.pBatchReqs) return terrno;

×

4159

    SVDropTbBatchReq batchReq = {0};

×

4160

    int32_t          code = createDropTbBatchReq(&req, &batchReq);

×

4161

    if (TSDB_CODE_SUCCESS != code) return code;

×

4162

    if (taosArrayPush(vgReqs.pBatchReqs, &batchReq) == NULL) {

×

4163

      taosArrayDestroy(batchReq.pArray);

×

4164

      return terrno;

×

4166

    if (taosHashPut(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId), &vgReqs, sizeof(vgReqs)) != 0) {

×

4167

      taosArrayDestroyEx(vgReqs.pBatchReqs, destroySVDropTbBatchReqs);

×

4168

      return terrno;

×

4171

    SVDropTbBatchReq batchReq = {0};

×

4172

    int32_t          code = createDropTbBatchReq(&req, &batchReq);

×

4173

    if (TSDB_CODE_SUCCESS != code) return code;

×

4174

    if (taosArrayPush(pVgReqs->pBatchReqs, &batchReq) == NULL) {

×

4175

      taosArrayDestroy(batchReq.pArray);

×

4176

      return terrno;

×

4179

  return 0;

×

4182

static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId) {

×

4183

  int32_t code = 0;

×

4185

  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);

×

4186

  if (!pVgObj) {

×

4187

    code = 0;

×

4188

    goto _end;

×

4190

  SVgroupInfo vgInfo = {.hashBegin = pVgObj->hashBegin,

×

4191

                        .hashEnd = pVgObj->hashEnd,

×

4192

                        .numOfTable = pVgObj->numOfTables,

×

4193

                        .vgId = pVgObj->vgId};

×

4194

  vgInfo.epSet = mndGetVgroupEpset(pMnode, pVgObj);

×

4195

  mndReleaseVgroup(pMnode, pVgObj);

×

4197

  for (int32_t i = 0; i < pTbs->size; ++i) {

×

4198

    SVDropTbReq *pTb = taosArrayGet(pTbs, i);

×

4199

    TAOS_CHECK_GOTO(mndDropTbAdd(pMnode, pCtx->pVgMap, &vgInfo, pTb->name, pTb->suid, pTb->igNotExists), NULL, _end);

×

4201

_end:

×

4202

  return code;

×

4205

static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pRsp) {

×

4206

  int32_t                 code = -1;

×

4207

  SDecoder                decoder = {0};

×

4208

  SMnode                 *pMnode = pRsp->info.node;

×

4209

  SVFetchTtlExpiredTbsRsp rsp = {0};

×

4210

  SMndDropTbsWithTsmaCtx *pCtx = NULL;

×

4211

  if (pRsp->code != TSDB_CODE_SUCCESS) {

×

4212

    code = pRsp->code;

×

4213

    goto _end;

×

4215

  if (pRsp->contLen == 0) {

×

4216

    code = 0;

×

4217

    goto _end;

×

4220

  tDecoderInit(&decoder, pRsp->pCont, pRsp->contLen);

×

4221

  code = tDecodeVFetchTtlExpiredTbsRsp(&decoder, &rsp);

×

4222

  if (code) goto _end;

×

4224

  code = mndInitDropTbsWithTsmaCtx(&pCtx);

×

4225

  if (code) goto _end;

×

4227

  code = mndDropTbForSingleVg(pMnode, pCtx, rsp.pExpiredTbs, rsp.vgId);

×

4228

  if (code) goto _end;

×

4229

  code = mndCreateDropTbsTxnPrepare(pRsp, pCtx);

×

4230

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

×

4231

_end:

×

4232

  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);

×

4233

  tDecoderClear(&decoder);

×

4234

  tFreeFetchTtlExpiredTbsRsp(&rsp);

×

4235

  TAOS_RETURN(code);

×

taosdata / TDengine / #3798

Source File
Press 'n' to go to next uncovered line, 'b' for previous

taosdata / TDengine / #3798

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous