mndSma.c 38.3 KB
Newer Older
S
sma  
Shengliang Guan 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#define _DEFAULT_SOURCE
#include "mndSma.h"
#include "mndDb.h"
#include "mndDnode.h"
#include "mndInfoSchema.h"
#include "mndMnode.h"
L
Liu Jicong 已提交
22
#include "mndPrivilege.h"
L
Liu Jicong 已提交
23
#include "mndScheduler.h"
S
sma  
Shengliang Guan 已提交
24
#include "mndShow.h"
S
Shengliang Guan 已提交
25
#include "mndStb.h"
S
sma  
Shengliang Guan 已提交
26
#include "mndStream.h"
S
sma  
Shengliang Guan 已提交
27 28 29
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
C
Cary Xu 已提交
30
#include "parser.h"
S
sma  
Shengliang Guan 已提交
31 32 33 34 35 36 37 38 39 40
#include "tname.h"

#define TSDB_SMA_VER_NUMBER   1
#define TSDB_SMA_RESERVE_SIZE 64

static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma);
static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw);
static int32_t  mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma);
static int32_t  mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSpSmatb);
static int32_t  mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew);
41 42
static int32_t  mndProcessCreateSmaReq(SRpcMsg *pReq);
static int32_t  mndProcessDropSmaReq(SRpcMsg *pReq);
S
Shengliang Guan 已提交
43
static int32_t  mndProcessGetSmaReq(SRpcMsg *pReq);
44
static int32_t  mndProcessGetTbSmaReq(SRpcMsg *pReq);
S
Shengliang Guan 已提交
45
static int32_t  mndRetrieveSma(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
S
sma  
Shengliang Guan 已提交
46
static void     mndCancelGetNextSma(SMnode *pMnode, void *pIter);
C
Cary Xu 已提交
47
static void     mndDestroySmaObj(SSmaObj *pSmaObj);
S
sma  
Shengliang Guan 已提交
48 49

int32_t mndInitSma(SMnode *pMnode) {
50 51 52 53 54 55 56 57 58
  SSdbTable table = {
      .sdbType = SDB_SMA,
      .keyType = SDB_KEY_BINARY,
      .encodeFp = (SdbEncodeFp)mndSmaActionEncode,
      .decodeFp = (SdbDecodeFp)mndSmaActionDecode,
      .insertFp = (SdbInsertFp)mndSmaActionInsert,
      .updateFp = (SdbUpdateFp)mndSmaActionUpdate,
      .deleteFp = (SdbDeleteFp)mndSmaActionDelete,
  };
S
sma  
Shengliang Guan 已提交
59

60 61
  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_SMA, mndProcessCreateSmaReq);
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_SMA, mndProcessDropSmaReq);
62 63
  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_SMA_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_SMA_RSP, mndTransProcessRsp);
S
Shengliang Guan 已提交
64
  mndSetMsgHandle(pMnode, TDMT_MND_GET_INDEX, mndProcessGetSmaReq);
D
dapan1121 已提交
65
  mndSetMsgHandle(pMnode, TDMT_MND_GET_TABLE_INDEX, mndProcessGetTbSmaReq);
S
sma  
Shengliang Guan 已提交
66 67 68 69 70 71 72 73 74 75 76

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_INDEX, mndRetrieveSma);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_INDEX, mndCancelGetNextSma);
  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupSma(SMnode *pMnode) {}

static SSdbRaw *mndSmaActionEncode(SSmaObj *pSma) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;

L
Liu Jicong 已提交
77 78
  int32_t size =
      sizeof(SSmaObj) + pSma->exprLen + pSma->tagsFilterLen + pSma->sqlLen + pSma->astLen + TSDB_SMA_RESERVE_SIZE;
S
sma  
Shengliang Guan 已提交
79 80 81 82 83 84 85
  SSdbRaw *pRaw = sdbAllocRaw(SDB_SMA, TSDB_SMA_VER_NUMBER, size);
  if (pRaw == NULL) goto _OVER;

  int32_t dataPos = 0;
  SDB_SET_BINARY(pRaw, dataPos, pSma->name, TSDB_TABLE_FNAME_LEN, _OVER)
  SDB_SET_BINARY(pRaw, dataPos, pSma->stb, TSDB_TABLE_FNAME_LEN, _OVER)
  SDB_SET_BINARY(pRaw, dataPos, pSma->db, TSDB_DB_FNAME_LEN, _OVER)
C
Cary Xu 已提交
86
  SDB_SET_BINARY(pRaw, dataPos, pSma->dstTbName, TSDB_DB_FNAME_LEN, _OVER)
S
sma  
Shengliang Guan 已提交
87 88 89 90
  SDB_SET_INT64(pRaw, dataPos, pSma->createdTime, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pSma->uid, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pSma->stbUid, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pSma->dbUid, _OVER)
C
Cary Xu 已提交
91
  SDB_SET_INT64(pRaw, dataPos, pSma->dstTbUid, _OVER)
S
sma  
Shengliang Guan 已提交
92 93 94
  SDB_SET_INT8(pRaw, dataPos, pSma->intervalUnit, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pSma->slidingUnit, _OVER)
  SDB_SET_INT8(pRaw, dataPos, pSma->timezone, _OVER)
S
sma  
Shengliang Guan 已提交
95
  SDB_SET_INT32(pRaw, dataPos, pSma->dstVgId, _OVER)
S
sma  
Shengliang Guan 已提交
96 97 98 99 100
  SDB_SET_INT64(pRaw, dataPos, pSma->interval, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pSma->offset, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pSma->sliding, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pSma->exprLen, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pSma->tagsFilterLen, _OVER)
S
sma  
Shengliang Guan 已提交
101 102
  SDB_SET_INT32(pRaw, dataPos, pSma->sqlLen, _OVER)
  SDB_SET_INT32(pRaw, dataPos, pSma->astLen, _OVER)
103

S
sma  
Shengliang Guan 已提交
104 105 106 107 108 109 110 111 112 113 114 115
  if (pSma->exprLen > 0) {
    SDB_SET_BINARY(pRaw, dataPos, pSma->expr, pSma->exprLen, _OVER)
  }
  if (pSma->tagsFilterLen > 0) {
    SDB_SET_BINARY(pRaw, dataPos, pSma->tagsFilter, pSma->tagsFilterLen, _OVER)
  }
  if (pSma->sqlLen > 0) {
    SDB_SET_BINARY(pRaw, dataPos, pSma->sql, pSma->sqlLen, _OVER)
  }
  if (pSma->astLen > 0) {
    SDB_SET_BINARY(pRaw, dataPos, pSma->ast, pSma->astLen, _OVER)
  }
S
sma  
Shengliang Guan 已提交
116 117 118

  SDB_SET_RESERVE(pRaw, dataPos, TSDB_SMA_RESERVE_SIZE, _OVER)
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
119

S
sma  
Shengliang Guan 已提交
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154
  terrno = 0;

_OVER:
  if (terrno != 0) {
    mError("sma:%s, failed to encode to raw:%p since %s", pSma->name, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
  }

  mTrace("sma:%s, encode to raw:%p, row:%p", pSma->name, pRaw, pSma);
  return pRaw;
}

static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;

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

  if (sver != TSDB_SMA_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
    goto _OVER;
  }

  SSdbRow *pRow = sdbAllocRow(sizeof(SSmaObj));
  if (pRow == NULL) goto _OVER;

  SSmaObj *pSma = sdbGetRowObj(pRow);
  if (pSma == NULL) goto _OVER;

  int32_t dataPos = 0;

  SDB_GET_BINARY(pRaw, dataPos, pSma->name, TSDB_TABLE_FNAME_LEN, _OVER)
  SDB_GET_BINARY(pRaw, dataPos, pSma->stb, TSDB_TABLE_FNAME_LEN, _OVER)
  SDB_GET_BINARY(pRaw, dataPos, pSma->db, TSDB_DB_FNAME_LEN, _OVER)
C
Cary Xu 已提交
155
  SDB_GET_BINARY(pRaw, dataPos, pSma->dstTbName, TSDB_DB_FNAME_LEN, _OVER)
S
sma  
Shengliang Guan 已提交
156 157 158 159
  SDB_GET_INT64(pRaw, dataPos, &pSma->createdTime, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pSma->uid, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pSma->stbUid, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pSma->dbUid, _OVER)
C
Cary Xu 已提交
160
  SDB_GET_INT64(pRaw, dataPos, &pSma->dstTbUid, _OVER)
S
sma  
Shengliang Guan 已提交
161 162 163
  SDB_GET_INT8(pRaw, dataPos, &pSma->intervalUnit, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pSma->slidingUnit, _OVER)
  SDB_GET_INT8(pRaw, dataPos, &pSma->timezone, _OVER)
S
sma  
Shengliang Guan 已提交
164
  SDB_GET_INT32(pRaw, dataPos, &pSma->dstVgId, _OVER)
S
sma  
Shengliang Guan 已提交
165 166 167 168 169
  SDB_GET_INT64(pRaw, dataPos, &pSma->interval, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pSma->offset, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pSma->sliding, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pSma->exprLen, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pSma->tagsFilterLen, _OVER)
S
sma  
Shengliang Guan 已提交
170 171
  SDB_GET_INT32(pRaw, dataPos, &pSma->sqlLen, _OVER)
  SDB_GET_INT32(pRaw, dataPos, &pSma->astLen, _OVER)
S
sma  
Shengliang Guan 已提交
172

S
sma  
Shengliang Guan 已提交
173
  if (pSma->exprLen > 0) {
wafwerar's avatar
wafwerar 已提交
174
    pSma->expr = taosMemoryCalloc(pSma->exprLen, 1);
S
sma  
Shengliang Guan 已提交
175 176
    if (pSma->expr == NULL) goto _OVER;
    SDB_GET_BINARY(pRaw, dataPos, pSma->expr, pSma->exprLen, _OVER)
S
sma  
Shengliang Guan 已提交
177 178
  }

S
sma  
Shengliang Guan 已提交
179
  if (pSma->tagsFilterLen > 0) {
wafwerar's avatar
wafwerar 已提交
180
    pSma->tagsFilter = taosMemoryCalloc(pSma->tagsFilterLen, 1);
S
sma  
Shengliang Guan 已提交
181 182 183
    if (pSma->tagsFilter == NULL) goto _OVER;
    SDB_GET_BINARY(pRaw, dataPos, pSma->tagsFilter, pSma->tagsFilterLen, _OVER)
  }
S
sma  
Shengliang Guan 已提交
184

S
sma  
Shengliang Guan 已提交
185
  if (pSma->sqlLen > 0) {
wafwerar's avatar
wafwerar 已提交
186
    pSma->sql = taosMemoryCalloc(pSma->sqlLen, 1);
S
sma  
Shengliang Guan 已提交
187 188 189 190 191
    if (pSma->sql == NULL) goto _OVER;
    SDB_GET_BINARY(pRaw, dataPos, pSma->sql, pSma->sqlLen, _OVER)
  }

  if (pSma->astLen > 0) {
wafwerar's avatar
wafwerar 已提交
192
    pSma->ast = taosMemoryCalloc(pSma->astLen, 1);
S
sma  
Shengliang Guan 已提交
193 194 195 196 197
    if (pSma->ast == NULL) goto _OVER;
    SDB_GET_BINARY(pRaw, dataPos, pSma->ast, pSma->astLen, _OVER)
  }

  SDB_GET_RESERVE(pRaw, dataPos, TSDB_SMA_RESERVE_SIZE, _OVER)
198

S
sma  
Shengliang Guan 已提交
199 200 201 202 203
  terrno = 0;

_OVER:
  if (terrno != 0) {
    mError("sma:%s, failed to decode from raw:%p since %s", pSma->name, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
204 205
    taosMemoryFreeClear(pSma->expr);
    taosMemoryFreeClear(pSma->tagsFilter);
206 207
    taosMemoryFreeClear(pSma->sql);
    taosMemoryFreeClear(pSma->ast);
wafwerar's avatar
wafwerar 已提交
208
    taosMemoryFreeClear(pRow);
S
sma  
Shengliang Guan 已提交
209 210 211 212 213 214 215 216 217 218 219 220 221 222
    return NULL;
  }

  mTrace("sma:%s, decode from raw:%p, row:%p", pSma->name, pRaw, pSma);
  return pRow;
}

static int32_t mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma) {
  mTrace("sma:%s, perform insert action, row:%p", pSma->name, pSma);
  return 0;
}

static int32_t mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSma) {
  mTrace("sma:%s, perform delete action, row:%p", pSma->name, pSma);
wafwerar's avatar
wafwerar 已提交
223 224
  taosMemoryFreeClear(pSma->tagsFilter);
  taosMemoryFreeClear(pSma->expr);
225 226
  taosMemoryFreeClear(pSma->sql);
  taosMemoryFreeClear(pSma->ast);
S
sma  
Shengliang Guan 已提交
227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259
  return 0;
}

static int32_t mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew) {
  mTrace("sma:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
  return 0;
}

SSmaObj *mndAcquireSma(SMnode *pMnode, char *smaName) {
  SSdb    *pSdb = pMnode->pSdb;
  SSmaObj *pSma = sdbAcquire(pSdb, SDB_SMA, smaName);
  if (pSma == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
    terrno = TSDB_CODE_MND_SMA_NOT_EXIST;
  }
  return pSma;
}

void mndReleaseSma(SMnode *pMnode, SSmaObj *pSma) {
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pSma);
}

SDbObj *mndAcquireDbBySma(SMnode *pMnode, const char *smaName) {
  SName name = {0};
  tNameFromString(&name, smaName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);

  char db[TSDB_TABLE_FNAME_LEN] = {0};
  tNameGetFullDbName(&name, db);

  return mndAcquireDb(pMnode, db);
}

static void *mndBuildVCreateSmaReq(SMnode *pMnode, SVgObj *pVgroup, SSmaObj *pSma, int32_t *pContLen) {
260 261 262
  SEncoder encoder = {0};
  int32_t  contLen = 0;
  SName    name = {0};
S
sma  
Shengliang Guan 已提交
263 264 265
  tNameFromString(&name, pSma->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);

  SVCreateTSmaReq req = {0};
266 267 268 269 270 271 272 273 274
  req.version = 0;
  req.intervalUnit = pSma->intervalUnit;
  req.slidingUnit = pSma->slidingUnit;
  req.timezoneInt = pSma->timezone;
  tstrncpy(req.indexName, (char *)tNameGetTableName(&name), TSDB_INDEX_NAME_LEN);
  req.exprLen = pSma->exprLen;
  req.tagsFilterLen = pSma->tagsFilterLen;
  req.indexUid = pSma->uid;
  req.tableUid = pSma->stbUid;
C
Cary Xu 已提交
275 276
  req.dstVgId = pSma->dstVgId;
  req.dstTbUid = pSma->dstTbUid;
277 278 279 280 281
  req.interval = pSma->interval;
  req.offset = pSma->offset;
  req.sliding = pSma->sliding;
  req.expr = pSma->expr;
  req.tagsFilter = pSma->tagsFilter;
C
Cary Xu 已提交
282 283 284 285
  req.schemaRow = pSma->schemaRow;
  req.schemaTag = pSma->schemaTag;
  req.dstTbName = pSma->dstTbName;

286 287 288 289 290 291 292 293
  // get length
  int32_t ret = 0;
  tEncodeSize(tEncodeSVCreateTSmaReq, &req, contLen, ret);
  if (ret < 0) {
    return NULL;
  }
  contLen += sizeof(SMsgHead);

wafwerar's avatar
wafwerar 已提交
294
  SMsgHead *pHead = taosMemoryMalloc(contLen);
S
sma  
Shengliang Guan 已提交
295 296 297 298 299 300 301 302 303
  if (pHead == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

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

  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
304 305 306 307 308 309 310 311
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
  if (tEncodeSVCreateTSmaReq(&encoder, &req) < 0) {
    taosMemoryFreeClear(pHead);
    tEncoderClear(&encoder);
    return NULL;
  }

  tEncoderClear(&encoder);
S
sma  
Shengliang Guan 已提交
312 313 314 315 316 317

  *pContLen = contLen;
  return pHead;
}

static void *mndBuildVDropSmaReq(SMnode *pMnode, SVgObj *pVgroup, SSmaObj *pSma, int32_t *pContLen) {
L
Liu Jicong 已提交
318 319 320
  SEncoder encoder = {0};
  int32_t  contLen;
  SName    name = {0};
S
sma  
Shengliang Guan 已提交
321 322 323 324 325 326
  tNameFromString(&name, pSma->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);

  SVDropTSmaReq req = {0};
  req.indexUid = pSma->uid;
  tstrncpy(req.indexName, (char *)tNameGetTableName(&name), TSDB_INDEX_NAME_LEN);

327 328 329 330 331 332 333 334 335
  // get length
  int32_t ret = 0;
  tEncodeSize(tEncodeSVDropTSmaReq, &req, contLen, ret);
  if (ret < 0) {
    return NULL;
  }

  contLen += sizeof(SMsgHead);

wafwerar's avatar
wafwerar 已提交
336
  SMsgHead *pHead = taosMemoryMalloc(contLen);
S
sma  
Shengliang Guan 已提交
337 338 339 340 341 342 343 344 345
  if (pHead == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

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

  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
346 347 348 349 350 351 352 353
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));

  if (tEncodeSVDropTSmaReq(&encoder, &req) < 0) {
    taosMemoryFreeClear(pHead);
    tEncoderClear(&encoder);
    return NULL;
  }
  tEncoderClear(&encoder);
S
sma  
Shengliang Guan 已提交
354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376

  *pContLen = contLen;
  return pHead;
}

static int32_t mndSetCreateSmaRedoLogs(SMnode *pMnode, STrans *pTrans, SSmaObj *pSma) {
  SSdbRaw *pRedoRaw = mndSmaActionEncode(pSma);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING) != 0) return -1;

  return 0;
}

static int32_t mndSetCreateSmaCommitLogs(SMnode *pMnode, STrans *pTrans, SSmaObj *pSma) {
  SSdbRaw *pCommitRaw = mndSmaActionEncode(pSma);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;

  return 0;
}

S
Shengliang Guan 已提交
377 378 379 380
static int32_t mndSetCreateSmaVgroupRedoLogs(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup) {
  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
  if (pVgRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) return -1;
S
Shengliang Guan 已提交
381
  if (sdbSetRawStatus(pVgRaw, SDB_STATUS_CREATING) != 0) return -1;
S
Shengliang Guan 已提交
382 383 384 385 386 387 388 389 390 391 392
  return 0;
}

static int32_t mndSetCreateSmaVgroupCommitLogs(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup) {
  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
  if (pVgRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) return -1;
  if (sdbSetRawStatus(pVgRaw, SDB_STATUS_READY) != 0) return -1;
  return 0;
}

393 394 395 396 397
static int32_t mndSetUpdateSmaStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
  SStbObj stbObj = {0};
  taosRLockLatch(&pStb->lock);
  memcpy(&stbObj, pStb, sizeof(SStbObj));
  taosRUnLockLatch(&pStb->lock);
C
Cary Xu 已提交
398
  stbObj.numOfColumns = 0;
399
  stbObj.pColumns = NULL;
C
Cary Xu 已提交
400
  stbObj.numOfTags = 0;
401
  stbObj.pTags = NULL;
S
Shengliang Guan 已提交
402 403
  stbObj.numOfFuncs = 0;
  stbObj.pFuncs = NULL;
404 405 406 407 408 409 410 411 412 413 414 415
  stbObj.updateTime = taosGetTimestampMs();
  stbObj.lock = 0;
  stbObj.smaVer++;

  SSdbRaw *pCommitRaw = mndStbActionEncode(&stbObj);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;

  return 0;
}

C
Cary Xu 已提交
416 417
static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup,
                                                SSmaObj *pSma) {
S
Shengliang Guan 已提交
418 419 420 421 422 423 424 425 426
  SVnodeGid *pVgid = pVgroup->vnodeGid + 0;
  SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
  if (pDnode == NULL) return -1;

  STransAction action = {0};
  action.epSet = mndGetDnodeEpset(pDnode);
  mndReleaseDnode(pMnode, pDnode);

  // todo add sma info here
C
Cary Xu 已提交
427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450
  SNode *pAst = NULL;
  if (nodesStringToNode(pSma->ast, &pAst) < 0) {
    return -1;
  }
  if (qExtractResultSchema(pAst, &pSma->schemaRow.nCols, &pSma->schemaRow.pSchema) != 0) {
    nodesDestroyNode(pAst);
    return -1;
  }
  nodesDestroyNode(pAst);
  pSma->schemaRow.version = 1;

  // TODO: the schemaTag generated by qExtractResultXXX later.
  pSma->schemaTag.nCols = 1;
  pSma->schemaTag.version = 1;
  pSma->schemaTag.pSchema = taosMemoryCalloc(1, sizeof(SSchema));
  if (!pSma->schemaTag.pSchema) {
    return -1;
  }
  pSma->schemaTag.pSchema[0].type = TSDB_DATA_TYPE_BIGINT;
  pSma->schemaTag.pSchema[0].bytes = TYPE_BYTES[TSDB_DATA_TYPE_BIGINT];
  pSma->schemaTag.pSchema[0].colId = pSma->schemaRow.nCols + PRIMARYKEY_TIMESTAMP_COL_ID;
  pSma->schemaTag.pSchema[0].flags = 0;
  snprintf(pSma->schemaTag.pSchema[0].name, TSDB_COL_NAME_LEN, "groupId");

C
Cary Xu 已提交
451 452 453 454
  int32_t smaContLen = 0;
  void   *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen);
  if (pSmaReq == NULL) return -1;
  pVgroup->pTsma = pSmaReq;
S
Shengliang Guan 已提交
455 456

  int32_t contLen = 0;
S
Shengliang Guan 已提交
457
  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
C
Cary Xu 已提交
458
  taosMemoryFreeClear(pSmaReq);
S
Shengliang Guan 已提交
459 460 461 462 463 464 465 466 467 468 469 470 471 472 473
  if (pReq == NULL) return -1;

  action.pCont = pReq;
  action.contLen = contLen;
  action.msgType = TDMT_DND_CREATE_VNODE;
  action.acceptableCode = TSDB_CODE_NODE_ALREADY_DEPLOYED;

  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
    taosMemoryFree(pReq);
    return -1;
  }

  return 0;
}

C
Cary Xu 已提交
474 475 476 477 478 479 480
static void mndDestroySmaObj(SSmaObj *pSmaObj) {
  if (pSmaObj) {
    taosMemoryFreeClear(pSmaObj->schemaRow.pSchema);
    taosMemoryFreeClear(pSmaObj->schemaTag.pSchema);
  }
}

481 482
static int32_t mndCreateSma(SMnode *pMnode, SRpcMsg *pReq, SMCreateSmaReq *pCreate, SDbObj *pDb, SStbObj *pStb,
                            const char *streamName) {
S
sma  
Shengliang Guan 已提交
483 484 485
  SSmaObj smaObj = {0};
  memcpy(smaObj.name, pCreate->name, TSDB_TABLE_FNAME_LEN);
  memcpy(smaObj.stb, pStb->name, TSDB_TABLE_FNAME_LEN);
S
sma  
Shengliang Guan 已提交
486
  memcpy(smaObj.db, pDb->name, TSDB_DB_FNAME_LEN);
S
sma  
Shengliang Guan 已提交
487 488
  smaObj.createdTime = taosGetTimestampMs();
  smaObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
L
Liu Jicong 已提交
489
  ASSERT(smaObj.uid != 0);
C
Cary Xu 已提交
490
  char resultTbName[TSDB_TABLE_FNAME_LEN + 16] = {0};
L
Liu Jicong 已提交
491
  snprintf(resultTbName, TSDB_TABLE_FNAME_LEN + 16, "%s_td_tsma_rst_tb", pCreate->name);
C
Cary Xu 已提交
492 493
  memcpy(smaObj.dstTbName, resultTbName, TSDB_TABLE_FNAME_LEN);
  smaObj.dstTbUid = mndGenerateUid(smaObj.dstTbName, TSDB_TABLE_FNAME_LEN);
S
sma  
Shengliang Guan 已提交
494
  smaObj.stbUid = pStb->uid;
S
sma  
Shengliang Guan 已提交
495
  smaObj.dbUid = pStb->dbUid;
S
sma  
Shengliang Guan 已提交
496 497
  smaObj.intervalUnit = pCreate->intervalUnit;
  smaObj.slidingUnit = pCreate->slidingUnit;
498
#if 0
S
sma  
Shengliang Guan 已提交
499
  smaObj.timezone = pCreate->timezone;
500
#endif
501
  smaObj.timezone = tsTimezone;  // use timezone of server
S
sma  
Shengliang Guan 已提交
502 503 504 505 506
  smaObj.interval = pCreate->interval;
  smaObj.offset = pCreate->offset;
  smaObj.sliding = pCreate->sliding;
  smaObj.exprLen = pCreate->exprLen;
  smaObj.tagsFilterLen = pCreate->tagsFilterLen;
S
sma  
Shengliang Guan 已提交
507 508 509
  smaObj.sqlLen = pCreate->sqlLen;
  smaObj.astLen = pCreate->astLen;
  if (smaObj.exprLen > 0) {
C
Cary Xu 已提交
510
    smaObj.expr = pCreate->expr;
S
sma  
Shengliang Guan 已提交
511 512
  }
  if (smaObj.tagsFilterLen > 0) {
C
Cary Xu 已提交
513
    smaObj.tagsFilter = pCreate->tagsFilter;
S
sma  
Shengliang Guan 已提交
514 515
  }
  if (smaObj.sqlLen > 0) {
C
Cary Xu 已提交
516
    smaObj.sql = pCreate->sql;
S
sma  
Shengliang Guan 已提交
517 518
  }
  if (smaObj.astLen > 0) {
C
Cary Xu 已提交
519
    smaObj.ast = pCreate->ast;
S
sma  
Shengliang Guan 已提交
520
  }
S
sma  
Shengliang Guan 已提交
521

S
sma  
Shengliang Guan 已提交
522
  SStreamObj streamObj = {0};
523
  tstrncpy(streamObj.name, streamName, TSDB_STREAM_FNAME_LEN);
L
Liu Jicong 已提交
524
  tstrncpy(streamObj.sourceDb, pDb->name, TSDB_DB_FNAME_LEN);
D
dapan1121 已提交
525
  tstrncpy(streamObj.targetDb, streamObj.sourceDb, TSDB_DB_FNAME_LEN);
S
sma  
Shengliang Guan 已提交
526 527
  streamObj.createTime = taosGetTimestampMs();
  streamObj.updateTime = streamObj.createTime;
528
  streamObj.uid = mndGenerateUid(streamName, strlen(streamName));
L
Liu Jicong 已提交
529
  streamObj.sourceDbUid = pDb->uid;
S
Shengliang Guan 已提交
530
  streamObj.targetDbUid = pDb->uid;
S
sma  
Shengliang Guan 已提交
531
  streamObj.version = 1;
L
Liu Jicong 已提交
532
  streamObj.sql = strdup(pCreate->sql);
L
Liu Jicong 已提交
533
  streamObj.smaId = smaObj.uid;
C
Cary Xu 已提交
534 535 536
  streamObj.watermark = pCreate->watermark;
  streamObj.trigger = STREAM_TRIGGER_WINDOW_CLOSE;
  streamObj.triggerParam = pCreate->maxDelay;
L
Liu Jicong 已提交
537
  streamObj.ast = strdup(smaObj.ast);
L
Liu Jicong 已提交
538

C
Cary Xu 已提交
539 540 541 542 543 544 545 546 547
  // check the maxDelay
  if (streamObj.triggerParam < TSDB_MIN_ROLLUP_MAX_DELAY) {
    int64_t msInterval = convertTimeFromPrecisionToUnit(pCreate->interval, pDb->cfg.precision, TIME_UNIT_MILLISECOND);
    streamObj.triggerParam = msInterval > TSDB_MIN_ROLLUP_MAX_DELAY ? msInterval : TSDB_MIN_ROLLUP_MAX_DELAY;
  }
  if (streamObj.triggerParam > TSDB_MAX_ROLLUP_MAX_DELAY) {
    streamObj.triggerParam = TSDB_MAX_ROLLUP_MAX_DELAY;
  }

L
Liu Jicong 已提交
548 549 550 551 552
  if (mndAllocSmaVgroup(pMnode, pDb, &streamObj.fixedSinkVg) != 0) {
    mError("sma:%s, failed to create since %s", smaObj.name, terrstr());
    return -1;
  }
  smaObj.dstVgId = streamObj.fixedSinkVg.vgId;
S
Shengliang Guan 已提交
553
  streamObj.fixedSinkVgId = smaObj.dstVgId;
S
sma  
Shengliang Guan 已提交
554

L
Liu Jicong 已提交
555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586
  SNode *pAst = NULL;
  if (nodesStringToNode(streamObj.ast, &pAst) < 0) {
    ASSERT(0);
    return -1;
  }

  // extract output schema from ast
  if (qExtractResultSchema(pAst, (int32_t *)&streamObj.outputSchema.nCols, &streamObj.outputSchema.pSchema) != 0) {
    ASSERT(0);
    return -1;
  }

  SQueryPlan  *pPlan = NULL;
  SPlanContext cxt = {
      .pAstRoot = pAst,
      .topicQuery = false,
      .streamQuery = true,
      .triggerType = streamObj.trigger,
      .watermark = streamObj.watermark,
  };

  if (qCreateQueryPlan(&cxt, &pPlan, NULL) < 0) {
    ASSERT(0);
    return -1;
  }

  // save physcial plan
  if (nodesNodeToString((SNode *)pPlan, false, &streamObj.physicalPlan, NULL) != 0) {
    ASSERT(0);
    return -1;
  }
  if (pAst != NULL) nodesDestroyNode(pAst);
L
Liu Jicong 已提交
587
  nodesDestroyNode((SNode *)pPlan);
L
Liu Jicong 已提交
588

S
sma  
Shengliang Guan 已提交
589
  int32_t code = -1;
590
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "create-sma");
S
sma  
Shengliang Guan 已提交
591
  if (pTrans == NULL) goto _OVER;
592
  mndTransSetDbName(pTrans, pDb->name, NULL);
593
  mndTransSetSerial(pTrans);
594
  mInfo("trans:%d, used to create sma:%s stream:%s", pTrans->id, pCreate->name, streamObj.name);
S
sma  
Shengliang Guan 已提交
595 596

  if (mndSetCreateSmaRedoLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
L
Liu Jicong 已提交
597
  if (mndSetCreateSmaVgroupRedoLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER;
S
sma  
Shengliang Guan 已提交
598
  if (mndSetCreateSmaCommitLogs(pMnode, pTrans, &smaObj) != 0) goto _OVER;
L
Liu Jicong 已提交
599
  if (mndSetCreateSmaVgroupCommitLogs(pMnode, pTrans, &streamObj.fixedSinkVg) != 0) goto _OVER;
600
  if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
C
Cary Xu 已提交
601
  if (mndSetCreateSmaVgroupRedoActions(pMnode, pTrans, pDb, &streamObj.fixedSinkVg, &smaObj) != 0) goto _OVER;
602
  if (mndScheduleStream(pMnode, &streamObj) != 0) goto _OVER;
L
Liu Jicong 已提交
603
  if (mndPersistStream(pMnode, pTrans, &streamObj) != 0) goto _OVER;
S
sma  
Shengliang Guan 已提交
604 605
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;

606
  mInfo("sma:%s, uid:%" PRIi64 " create on stb:%" PRIi64 ", dstSuid:%" PRIi64 " dstTb:%s dstVg:%d", pCreate->name,
H
Hongze Cheng 已提交
607
        smaObj.uid, smaObj.stbUid, smaObj.dstTbUid, smaObj.dstTbName, smaObj.dstVgId);
C
Cary Xu 已提交
608

S
sma  
Shengliang Guan 已提交
609 610 611
  code = 0;

_OVER:
L
Liu Jicong 已提交
612
  tFreeStreamObj(&streamObj);
C
Cary Xu 已提交
613
  mndDestroySmaObj(&smaObj);
S
sma  
Shengliang Guan 已提交
614 615 616 617 618
  mndTransDrop(pTrans);
  return code;
}

static int32_t mndCheckCreateSmaReq(SMCreateSmaReq *pCreate) {
S
sma  
Shengliang Guan 已提交
619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640
  terrno = TSDB_CODE_MND_INVALID_SMA_OPTION;
  if (pCreate->name[0] == 0) return -1;
  if (pCreate->stb[0] == 0) return -1;
  if (pCreate->igExists < 0 || pCreate->igExists > 1) return -1;
  if (pCreate->intervalUnit < 0) return -1;
  if (pCreate->slidingUnit < 0) return -1;
  if (pCreate->timezone < 0) return -1;
  if (pCreate->interval < 0) return -1;
  if (pCreate->offset < 0) return -1;
  if (pCreate->sliding < 0) return -1;
  if (pCreate->exprLen < 0) return -1;
  if (pCreate->tagsFilterLen < 0) return -1;
  if (pCreate->sqlLen < 0) return -1;
  if (pCreate->astLen < 0) return -1;
  if (pCreate->exprLen != 0 && strlen(pCreate->expr) + 1 != pCreate->exprLen) return -1;
  if (pCreate->tagsFilterLen != 0 && strlen(pCreate->tagsFilter) + 1 != pCreate->tagsFilterLen) return -1;
  if (pCreate->sqlLen != 0 && strlen(pCreate->sql) + 1 != pCreate->sqlLen) return -1;
  if (pCreate->astLen != 0 && strlen(pCreate->ast) + 1 != pCreate->astLen) return -1;

  SName smaName = {0};
  if (tNameFromString(&smaName, pCreate->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE) < 0) return -1;
  if (*(char *)tNameGetTableName(&smaName) == 0) return -1;
S
sma  
Shengliang Guan 已提交
641

S
sma  
Shengliang Guan 已提交
642
  terrno = 0;
S
sma  
Shengliang Guan 已提交
643 644 645
  return 0;
}

646 647 648 649 650 651 652 653
static void mndGetStreamNameFromSmaName(char *streamName, char *smaName) {
  SName n;
  tNameFromString(&n, smaName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
  streamName[0] = '1';
  streamName[1] = '.';
  strcpy(streamName + 2, tNameGetTableName(&n));
}

654
static int32_t mndProcessCreateSmaReq(SRpcMsg *pReq) {
S
Shengliang Guan 已提交
655
  SMnode        *pMnode = pReq->info.node;
S
sma  
Shengliang Guan 已提交
656 657 658
  int32_t        code = -1;
  SStbObj       *pStb = NULL;
  SSmaObj       *pSma = NULL;
S
sma  
Shengliang Guan 已提交
659
  SStreamObj    *pStream = NULL;
S
sma  
Shengliang Guan 已提交
660 661 662
  SDbObj        *pDb = NULL;
  SMCreateSmaReq createReq = {0};

S
Shengliang Guan 已提交
663
  if (tDeserializeSMCreateSmaReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
S
sma  
Shengliang Guan 已提交
664 665 666 667
    terrno = TSDB_CODE_INVALID_MSG;
    goto _OVER;
  }

668
  mInfo("sma:%s, start to create", createReq.name);
S
sma  
Shengliang Guan 已提交
669 670 671 672 673 674 675 676 677
  if (mndCheckCreateSmaReq(&createReq) != 0) {
    goto _OVER;
  }

  pStb = mndAcquireStb(pMnode, createReq.stb);
  if (pStb == NULL) {
    mError("sma:%s, failed to create since stb:%s not exist", createReq.name, createReq.stb);
    goto _OVER;
  }
L
Liu Jicong 已提交
678

679 680 681 682
  char streamName[TSDB_TABLE_FNAME_LEN] = {0};
  mndGetStreamNameFromSmaName(streamName, createReq.name);

  pStream = mndAcquireStream(pMnode, streamName);
S
sma  
Shengliang Guan 已提交
683
  if (pStream != NULL) {
684
    mError("sma:%s, failed to create since stream:%s already exist", createReq.name, streamName);
C
Cary Xu 已提交
685
    terrno = TSDB_CODE_MND_STREAM_ALREADY_EXIST;
S
sma  
Shengliang Guan 已提交
686 687
    goto _OVER;
  }
S
sma  
Shengliang Guan 已提交
688 689 690 691

  pSma = mndAcquireSma(pMnode, createReq.name);
  if (pSma != NULL) {
    if (createReq.igExists) {
692
      mInfo("sma:%s, already exist in sma:%s, ignore exist is set", createReq.name, pSma->name);
S
sma  
Shengliang Guan 已提交
693 694 695 696 697 698 699 700 701 702 703 704 705 706
      code = 0;
      goto _OVER;
    } else {
      terrno = TSDB_CODE_MND_SMA_ALREADY_EXIST;
      goto _OVER;
    }
  }

  pDb = mndAcquireDbBySma(pMnode, createReq.name);
  if (pDb == NULL) {
    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
    goto _OVER;
  }

707
  if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb) != 0) {
S
sma  
Shengliang Guan 已提交
708 709 710
    goto _OVER;
  }

711
  code = mndCreateSma(pMnode, pReq, &createReq, pDb, pStb, streamName);
S
Shengliang Guan 已提交
712
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
S
sma  
Shengliang Guan 已提交
713 714

_OVER:
S
Shengliang Guan 已提交
715
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
C
Cary Xu 已提交
716
    mError("sma:%s, failed to create since %s", createReq.name, terrstr());
S
sma  
Shengliang Guan 已提交
717 718 719 720
  }

  mndReleaseStb(pMnode, pStb);
  mndReleaseSma(pMnode, pSma);
S
sma  
Shengliang Guan 已提交
721
  mndReleaseStream(pMnode, pStream);
S
sma  
Shengliang Guan 已提交
722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745
  mndReleaseDb(pMnode, pDb);
  tFreeSMCreateSmaReq(&createReq);

  return code;
}

static int32_t mndSetDropSmaRedoLogs(SMnode *pMnode, STrans *pTrans, SSmaObj *pSma) {
  SSdbRaw *pRedoRaw = mndSmaActionEncode(pSma);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING) != 0) return -1;

  return 0;
}

static int32_t mndSetDropSmaCommitLogs(SMnode *pMnode, STrans *pTrans, SSmaObj *pSma) {
  SSdbRaw *pCommitRaw = mndSmaActionEncode(pSma);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;

  return 0;
}

L
Liu Jicong 已提交
746
static int32_t mndSetDropSmaVgroupRedoLogs(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup) {
S
Shengliang Guan 已提交
747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789
  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
  if (pVgRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) return -1;
  if (sdbSetRawStatus(pVgRaw, SDB_STATUS_DROPPING) != 0) return -1;

  return 0;
}

static int32_t mndSetDropSmaVgroupCommitLogs(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup) {
  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
  if (pVgRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) return -1;
  if (sdbSetRawStatus(pVgRaw, SDB_STATUS_DROPPED) != 0) return -1;

  return 0;
}

static int32_t mndSetDropSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
  SVnodeGid *pVgid = pVgroup->vnodeGid + 0;
  SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
  if (pDnode == NULL) return -1;

  STransAction action = {0};
  action.epSet = mndGetDnodeEpset(pDnode);
  mndReleaseDnode(pMnode, pDnode);

  int32_t contLen = 0;
  void   *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
  if (pReq == NULL) return -1;

  action.pCont = pReq;
  action.contLen = contLen;
  action.msgType = TDMT_DND_DROP_VNODE;
  action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;

  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
    taosMemoryFree(pReq);
    return -1;
  }

  return 0;
}

S
Shengliang Guan 已提交
790
static int32_t mndDropSma(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SSmaObj *pSma) {
791 792 793 794
  int32_t  code = -1;
  SVgObj  *pVgroup = NULL;
  SStbObj *pStb = NULL;
  STrans  *pTrans = NULL;
S
Shengliang Guan 已提交
795 796 797 798

  pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
  if (pVgroup == NULL) goto _OVER;

799 800 801
  pStb = mndAcquireStb(pMnode, pSma->stb);
  if (pStb == NULL) goto _OVER;

802
  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "drop-sma");
S
sma  
Shengliang Guan 已提交
803 804
  if (pTrans == NULL) goto _OVER;

805
  mInfo("trans:%d, used to drop sma:%s", pTrans->id, pSma->name);
806
  mndTransSetDbName(pTrans, pDb->name, NULL);
C
Cary Xu 已提交
807
  mndTransSetSerial(pTrans);
S
sma  
Shengliang Guan 已提交
808

809 810 811 812
  char streamName[TSDB_TABLE_FNAME_LEN] = {0};
  mndGetStreamNameFromSmaName(streamName, pSma->name);

  SStreamObj *pStream = mndAcquireStream(pMnode, streamName);
L
Liu Jicong 已提交
813 814 815 816 817 818 819
  if (pStream == NULL || pStream->smaId != pSma->uid) {
    sdbRelease(pMnode->pSdb, pStream);
    goto _OVER;
  } else {
    if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
      mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
      sdbRelease(pMnode->pSdb, pStream);
820
      ASSERT(0);
L
Liu Jicong 已提交
821 822 823 824 825 826
      goto _OVER;
    }

    // drop stream
    if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
      sdbRelease(pMnode->pSdb, pStream);
827
      ASSERT(0);
L
Liu Jicong 已提交
828 829 830
      goto _OVER;
    }
  }
S
sma  
Shengliang Guan 已提交
831
  if (mndSetDropSmaRedoLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
S
Shengliang Guan 已提交
832
  if (mndSetDropSmaVgroupRedoLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
S
sma  
Shengliang Guan 已提交
833
  if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
S
Shengliang Guan 已提交
834
  if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
835
  if (mndSetUpdateSmaStbCommitLogs(pMnode, pTrans, pStb) != 0) goto _OVER;
S
Shengliang Guan 已提交
836
  if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
S
sma  
Shengliang Guan 已提交
837 838 839 840 841 842
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;

  code = 0;

_OVER:
  mndTransDrop(pTrans);
843
  mndReleaseStream(pMnode, pStream);
S
Shengliang Guan 已提交
844
  mndReleaseVgroup(pMnode, pVgroup);
845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860
  mndReleaseStb(pMnode, pStb);
  return code;
}

int32_t mndDropSmasByStb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
  SSdb    *pSdb = pMnode->pSdb;
  SSmaObj *pSma = NULL;
  void    *pIter = NULL;
  SVgObj  *pVgroup = NULL;
  int32_t  code = -1;

  while (1) {
    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
    if (pIter == NULL) break;

    if (pSma->stbUid == pStb->uid) {
C
Cary Xu 已提交
861
      mndTransSetSerial(pTrans);
862 863
      pVgroup = mndAcquireVgroup(pMnode, pSma->dstVgId);
      if (pVgroup == NULL) goto _OVER;
S
Shengliang Guan 已提交
864

865 866 867 868
      char streamName[TSDB_TABLE_FNAME_LEN] = {0};
      mndGetStreamNameFromSmaName(streamName, pSma->name);

      SStreamObj *pStream = mndAcquireStream(pMnode, streamName);
S
Shengliang Guan 已提交
869 870 871 872 873 874 875 876 877
      if (pStream != NULL && pStream->smaId == pSma->uid) {
        if (mndDropStreamTasks(pMnode, pTrans, pStream) < 0) {
          mError("stream:%s, failed to drop task since %s", pStream->name, terrstr());
          goto _OVER;
        }
        if (mndPersistDropStreamLog(pMnode, pTrans, pStream) < 0) {
          goto _OVER;
        }
      }
878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893
      if (mndSetDropSmaVgroupCommitLogs(pMnode, pTrans, pVgroup) != 0) goto _OVER;
      if (mndSetDropSmaVgroupRedoActions(pMnode, pTrans, pDb, pVgroup) != 0) goto _OVER;
      if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) goto _OVER;
      mndReleaseVgroup(pMnode, pVgroup);
      pVgroup = NULL;
    }

    sdbRelease(pSdb, pSma);
  }

  code = 0;

_OVER:
  sdbCancelFetch(pSdb, pIter);
  sdbRelease(pSdb, pSma);
  mndReleaseVgroup(pMnode, pVgroup);
S
sma  
Shengliang Guan 已提交
894 895 896
  return code;
}

897
int32_t mndDropSmasByDb(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
S
Shengliang Guan 已提交
898 899
  SSdb *pSdb = pMnode->pSdb;
  void *pIter = NULL;
900 901

  while (1) {
S
Shengliang Guan 已提交
902
    SSmaObj *pSma = NULL;
903 904 905 906
    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
    if (pIter == NULL) break;

    if (pSma->dbUid == pDb->uid) {
S
Shengliang Guan 已提交
907 908 909 910
      if (mndSetDropSmaCommitLogs(pMnode, pTrans, pSma) != 0) {
        sdbRelease(pSdb, pSma);
        sdbCancelFetch(pSdb, pSma);
        return -1;
S
Shengliang Guan 已提交
911
      }
912 913 914 915 916
    }

    sdbRelease(pSdb, pSma);
  }

S
Shengliang Guan 已提交
917
  return 0;
918 919 920
}

static int32_t mndProcessDropSmaReq(SRpcMsg *pReq) {
S
Shengliang Guan 已提交
921
  SMnode      *pMnode = pReq->info.node;
S
sma  
Shengliang Guan 已提交
922 923 924 925 926
  int32_t      code = -1;
  SDbObj      *pDb = NULL;
  SSmaObj     *pSma = NULL;
  SMDropSmaReq dropReq = {0};

S
Shengliang Guan 已提交
927
  if (tDeserializeSMDropSmaReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
S
sma  
Shengliang Guan 已提交
928 929 930 931
    terrno = TSDB_CODE_INVALID_MSG;
    goto _OVER;
  }

932
  mInfo("sma:%s, start to drop", dropReq.name);
S
sma  
Shengliang Guan 已提交
933 934 935 936

  pSma = mndAcquireSma(pMnode, dropReq.name);
  if (pSma == NULL) {
    if (dropReq.igNotExists) {
937
      mInfo("sma:%s, not exist, ignore not exist is set", dropReq.name);
S
sma  
Shengliang Guan 已提交
938 939 940 941 942 943 944 945 946 947 948 949 950 951
      code = 0;
      goto _OVER;
    } else {
      terrno = TSDB_CODE_MND_SMA_NOT_EXIST;
      goto _OVER;
    }
  }

  pDb = mndAcquireDbBySma(pMnode, dropReq.name);
  if (pDb == NULL) {
    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
    goto _OVER;
  }

952
  if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb) != 0) {
S
sma  
Shengliang Guan 已提交
953 954 955 956
    goto _OVER;
  }

  code = mndDropSma(pMnode, pReq, pDb, pSma);
S
Shengliang Guan 已提交
957
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
S
sma  
Shengliang Guan 已提交
958 959

_OVER:
S
Shengliang Guan 已提交
960
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
S
sma  
Shengliang Guan 已提交
961 962 963
    mError("sma:%s, failed to drop since %s", dropReq.name, terrstr());
  }

964
  mndReleaseSma(pMnode, pSma);
965
  mndReleaseDb(pMnode, pDb);
S
sma  
Shengliang Guan 已提交
966 967 968
  return code;
}

S
Shengliang Guan 已提交
969
static int32_t mndGetSma(SMnode *pMnode, SUserIndexReq *indexReq, SUserIndexRsp *rsp, bool *exist) {
L
Liu Jicong 已提交
970 971
  int32_t  code = -1;
  SSmaObj *pSma = NULL;
D
dapan1121 已提交
972

D
dapan1121 已提交
973
  pSma = mndAcquireSma(pMnode, indexReq->indexFName);
D
dapan1121 已提交
974 975 976 977 978 979 980 981 982 983
  if (pSma == NULL) {
    *exist = false;
    return 0;
  }

  memcpy(rsp->dbFName, pSma->db, sizeof(pSma->db));
  memcpy(rsp->tblFName, pSma->stb, sizeof(pSma->stb));
  strcpy(rsp->indexType, TSDB_INDEX_TYPE_SMA);

  SNodeList *pList = NULL;
L
Liu Jicong 已提交
984
  int32_t    extOffset = 0;
D
dapan1121 已提交
985 986 987 988 989
  code = nodesStringToList(pSma->expr, &pList);
  if (0 == code) {
    SNode *node = NULL;
    FOREACH(node, pList) {
      SFunctionNode *pFunc = (SFunctionNode *)node;
L
Liu Jicong 已提交
990 991
      extOffset += snprintf(rsp->indexExts + extOffset, sizeof(rsp->indexExts) - extOffset - 1, "%s%s",
                            (extOffset ? "," : ""), pFunc->functionName);
D
dapan1121 已提交
992 993 994 995 996 997
    }

    *exist = true;
  }

  mndReleaseSma(pMnode, pSma);
S
Shengliang Guan 已提交
998 999 1000
  return code;
}

D
dapan1121 已提交
1001
int32_t mndGetTableSma(SMnode *pMnode, char *tbFName, STableIndexRsp *rsp, bool *exist) {
1002 1003 1004 1005
  int32_t         code = 0;
  SSmaObj        *pSma = NULL;
  SSdb           *pSdb = pMnode->pSdb;
  void           *pIter = NULL;
D
dapan1121 已提交
1006 1007
  STableIndexInfo info;

1008
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
D
dapan1121 已提交
1009 1010 1011 1012 1013
  if (NULL == pStb) {
    *exist = false;
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
1014
  strcpy(rsp->dbFName, pStb->db);
D
dapan1121 已提交
1015
  strcpy(rsp->tbName, pStb->name + strlen(pStb->db) + 1);
D
dapan1121 已提交
1016 1017 1018 1019
  rsp->suid = pStb->uid;
  rsp->version = pStb->smaVer;
  mndReleaseStb(pMnode, pStb);

D
dapan1121 已提交
1020 1021 1022 1023
  while (1) {
    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
    if (pIter == NULL) break;

D
dapan1121 已提交
1024
    if (pSma->stb[0] != tbFName[0] || strcmp(pSma->stb, tbFName)) {
D
dapan1121 已提交
1025 1026 1027 1028 1029 1030 1031 1032 1033 1034
      continue;
    }

    info.intervalUnit = pSma->intervalUnit;
    info.slidingUnit = pSma->slidingUnit;
    info.interval = pSma->interval;
    info.offset = pSma->offset;
    info.sliding = pSma->sliding;
    info.dstTbUid = pSma->dstTbUid;
    info.dstVgId = pSma->dstVgId;
D
dapan1121 已提交
1035

1036
    SVgObj *pVg = mndAcquireVgroup(pMnode, pSma->dstVgId);
D
dapan1121 已提交
1037 1038 1039 1040 1041 1042
    if (pVg == NULL) {
      code = -1;
      sdbRelease(pSdb, pSma);
      return code;
    }
    info.epSet = mndGetVgroupEpset(pMnode, pVg);
1043

D
dapan1121 已提交
1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066
    info.expr = taosMemoryMalloc(pSma->exprLen + 1);
    if (info.expr == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      code = -1;
      sdbRelease(pSdb, pSma);
      return code;
    }

    memcpy(info.expr, pSma->expr, pSma->exprLen);
    info.expr[pSma->exprLen] = 0;

    if (NULL == taosArrayPush(rsp->pIndex, &info)) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      code = -1;
      taosMemoryFree(info.expr);
      sdbRelease(pSdb, pSma);
      return code;
    }

    *exist = true;

    sdbRelease(pSdb, pSma);
  }
1067

D
dapan1121 已提交
1068 1069 1070
  return code;
}

S
Shengliang Guan 已提交
1071
static int32_t mndProcessGetSmaReq(SRpcMsg *pReq) {
S
Shengliang Guan 已提交
1072
  SUserIndexReq indexReq = {0};
S
Shengliang Guan 已提交
1073
  SMnode       *pMnode = pReq->info.node;
S
Shengliang Guan 已提交
1074 1075 1076 1077
  int32_t       code = -1;
  SUserIndexRsp rsp = {0};
  bool          exist = false;

S
Shengliang Guan 已提交
1078
  if (tDeserializeSUserIndexReq(pReq->pCont, pReq->contLen, &indexReq) != 0) {
S
Shengliang Guan 已提交
1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102
    terrno = TSDB_CODE_INVALID_MSG;
    goto _OVER;
  }

  code = mndGetSma(pMnode, &indexReq, &rsp, &exist);
  if (code) {
    goto _OVER;
  }

  if (!exist) {
    // TODO GET INDEX FROM FULLTEXT
    code = -1;
    terrno = TSDB_CODE_MND_DB_INDEX_NOT_EXIST;
  } else {
    int32_t contLen = tSerializeSUserIndexRsp(NULL, 0, &rsp);
    void   *pRsp = rpcMallocCont(contLen);
    if (pRsp == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      code = -1;
      goto _OVER;
    }

    tSerializeSUserIndexRsp(pRsp, contLen, &rsp);

S
Shengliang Guan 已提交
1103 1104
    pReq->info.rsp = pRsp;
    pReq->info.rspLen = contLen;
S
Shengliang Guan 已提交
1105 1106 1107 1108 1109 1110 1111 1112

    code = 0;
  }

_OVER:
  if (code != 0) {
    mError("failed to get index %s since %s", indexReq.indexFName, terrstr());
  }
D
dapan1121 已提交
1113 1114 1115 1116

  return code;
}

D
dapan1121 已提交
1117 1118
static int32_t mndProcessGetTbSmaReq(SRpcMsg *pReq) {
  STableIndexReq indexReq = {0};
1119 1120
  SMnode        *pMnode = pReq->info.node;
  int32_t        code = -1;
D
dapan1121 已提交
1121
  STableIndexRsp rsp = {0};
1122
  bool           exist = false;
D
dapan1121 已提交
1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135

  if (tDeserializeSTableIndexReq(pReq->pCont, pReq->contLen, &indexReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    goto _OVER;
  }

  rsp.pIndex = taosArrayInit(10, sizeof(STableIndexInfo));
  if (NULL == rsp.pIndex) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    code = -1;
    goto _OVER;
  }

D
dapan1121 已提交
1136
  code = mndGetTableSma(pMnode, indexReq.tbFName, &rsp, &exist);
D
dapan1121 已提交
1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165
  if (code) {
    goto _OVER;
  }

  if (!exist) {
    code = -1;
    terrno = TSDB_CODE_MND_DB_INDEX_NOT_EXIST;
  } else {
    int32_t contLen = tSerializeSTableIndexRsp(NULL, 0, &rsp);
    void   *pRsp = rpcMallocCont(contLen);
    if (pRsp == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      code = -1;
      goto _OVER;
    }

    tSerializeSTableIndexRsp(pRsp, contLen, &rsp);

    pReq->info.rsp = pRsp;
    pReq->info.rspLen = contLen;

    code = 0;
  }

_OVER:
  if (code != 0) {
    mError("failed to get table index %s since %s", indexReq.tbFName, terrstr());
  }

1166
  tFreeSerializeSTableIndexRsp(&rsp);
D
dapan1121 已提交
1167 1168 1169
  return code;
}

S
Shengliang Guan 已提交
1170 1171
static int32_t mndRetrieveSma(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
  SMnode  *pMnode = pReq->info.node;
S
sma  
Shengliang Guan 已提交
1172 1173 1174 1175 1176
  SSdb    *pSdb = pMnode->pSdb;
  int32_t  numOfRows = 0;
  SSmaObj *pSma = NULL;
  int32_t  cols = 0;

1177 1178 1179 1180 1181
  SDbObj *pDb = NULL;
  if (strlen(pShow->db) > 0) {
    pDb = mndAcquireDb(pMnode, pShow->db);
    if (pDb == NULL) return 0;
  }
S
sma  
Shengliang Guan 已提交
1182 1183 1184 1185 1186

  while (numOfRows < rows) {
    pShow->pIter = sdbFetch(pSdb, SDB_SMA, pShow->pIter, (void **)&pSma);
    if (pShow->pIter == NULL) break;

1187
    if (NULL != pDb && pSma->dbUid != pDb->uid) {
S
sma  
Shengliang Guan 已提交
1188 1189 1190 1191 1192 1193 1194 1195
      sdbRelease(pSdb, pSma);
      continue;
    }

    cols = 0;

    SName smaName = {0};
    tNameFromString(&smaName, pSma->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
X
Xiaoyu Wang 已提交
1196 1197 1198
    char n1[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_TO_VARSTR(n1, (char *)tNameGetTableName(&smaName));

S
Shengliang Guan 已提交
1199
    char n2[TSDB_DB_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
1200
    STR_TO_VARSTR(n2, (char *)mndGetDbStr(pSma->db));
X
Xiaoyu Wang 已提交
1201

S
sma  
Shengliang Guan 已提交
1202 1203
    SName stbName = {0};
    tNameFromString(&stbName, pSma->stb, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
X
Xiaoyu Wang 已提交
1204 1205
    char n3[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_TO_VARSTR(n3, (char *)tNameGetTableName(&stbName));
1206

L
Liu Jicong 已提交
1207
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
X
Xiaoyu Wang 已提交
1208
    colDataAppend(pColInfo, numOfRows, (const char *)n1, false);
1209 1210

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
X
Xiaoyu Wang 已提交
1211 1212 1213 1214
    colDataAppend(pColInfo, numOfRows, (const char *)n2, false);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)n3, false);
S
sma  
Shengliang Guan 已提交
1215

S
Shengliang Guan 已提交
1216 1217 1218
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pSma->dstVgId, false);

S
Shengliang Guan 已提交
1219 1220 1221
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pSma->createdTime, false);

S
sma  
Shengliang Guan 已提交
1222 1223 1224 1225
    numOfRows++;
    sdbRelease(pSdb, pSma);
  }

X
Xiaoyu Wang 已提交
1226
  mndReleaseDb(pMnode, pDb);
1227
  pShow->numOfRows += numOfRows;
S
sma  
Shengliang Guan 已提交
1228 1229 1230 1231 1232 1233
  return numOfRows;
}

static void mndCancelGetNextSma(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
D
dapan1121 已提交
1234
}