ctgCache.c 67.8 KB
Newer Older
D
dapan1121 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
/*
 * 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/>.
 */

#include "catalogInt.h"
dengyihao's avatar
dengyihao 已提交
17
#include "query.h"
D
dapan1121 已提交
18
#include "systable.h"
dengyihao's avatar
dengyihao 已提交
19 20
#include "tname.h"
#include "trpc.h"
D
dapan1121 已提交
21

dengyihao's avatar
dengyihao 已提交
22 23 24 25 26 27 28 29 30 31 32
SCtgOperation gCtgCacheOperation[CTG_OP_MAX] = {{CTG_OP_UPDATE_VGROUP, "update vgInfo", ctgOpUpdateVgroup},
                                                {CTG_OP_UPDATE_TB_META, "update tbMeta", ctgOpUpdateTbMeta},
                                                {CTG_OP_DROP_DB_CACHE, "drop DB", ctgOpDropDbCache},
                                                {CTG_OP_DROP_DB_VGROUP, "drop DBVgroup", ctgOpDropDbVgroup},
                                                {CTG_OP_DROP_STB_META, "drop stbMeta", ctgOpDropStbMeta},
                                                {CTG_OP_DROP_TB_META, "drop tbMeta", ctgOpDropTbMeta},
                                                {CTG_OP_UPDATE_USER, "update user", ctgOpUpdateUser},
                                                {CTG_OP_UPDATE_VG_EPSET, "update epset", ctgOpUpdateEpset},
                                                {CTG_OP_UPDATE_TB_INDEX, "update tbIndex", ctgOpUpdateTbIndex},
                                                {CTG_OP_DROP_TB_INDEX, "drop tbIndex", ctgOpDropTbIndex},
                                                {CTG_OP_CLEAR_CACHE, "clear cache", ctgOpClearCache}};
D
dapan1121 已提交
33

D
dapan1121 已提交
34 35
int32_t ctgRLockVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
  CTG_LOCK(CTG_READ, &dbCache->vgCache.vgLock);
dengyihao's avatar
dengyihao 已提交
36

D
dapan1121 已提交
37
  if (dbCache->deleted) {
D
dapan1121 已提交
38
    CTG_UNLOCK(CTG_READ, &dbCache->vgCache.vgLock);
D
dapan1121 已提交
39

dengyihao's avatar
dengyihao 已提交
40 41
    ctgDebug("db is dropping, dbId:0x%" PRIx64, dbCache->dbId);

D
dapan1121 已提交
42 43 44 45
    *inCache = false;
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
46 47
  if (NULL == dbCache->vgCache.vgInfo) {
    CTG_UNLOCK(CTG_READ, &dbCache->vgCache.vgLock);
D
dapan1121 已提交
48 49

    *inCache = false;
dengyihao's avatar
dengyihao 已提交
50
    ctgDebug("db vgInfo is empty, dbId:0x%" PRIx64, dbCache->dbId);
D
dapan1121 已提交
51 52 53 54
    return TSDB_CODE_SUCCESS;
  }

  *inCache = true;
dengyihao's avatar
dengyihao 已提交
55

D
dapan1121 已提交
56 57 58
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
59 60
int32_t ctgWLockVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache) {
  CTG_LOCK(CTG_WRITE, &dbCache->vgCache.vgLock);
D
dapan1121 已提交
61 62

  if (dbCache->deleted) {
dengyihao's avatar
dengyihao 已提交
63
    ctgDebug("db is dropping, dbId:0x%" PRIx64, dbCache->dbId);
D
dapan1121 已提交
64
    CTG_UNLOCK(CTG_WRITE, &dbCache->vgCache.vgLock);
D
dapan1121 已提交
65 66 67 68 69 70
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
  }

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
71
void ctgRUnlockVgInfo(SCtgDBCache *dbCache) { CTG_UNLOCK(CTG_READ, &dbCache->vgCache.vgLock); }
D
dapan1121 已提交
72

dengyihao's avatar
dengyihao 已提交
73
void ctgWUnlockVgInfo(SCtgDBCache *dbCache) { CTG_UNLOCK(CTG_WRITE, &dbCache->vgCache.vgLock); }
D
dapan1121 已提交
74

dengyihao's avatar
dengyihao 已提交
75
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache) { CTG_UNLOCK(CTG_READ, &dbCache->dbLock); }
D
dapan1121 已提交
76

dengyihao's avatar
dengyihao 已提交
77
int32_t ctgAcquireDBCacheImpl(SCatalog *pCtg, const char *dbFName, SCtgDBCache **pCache, bool acquire) {
D
dapan1121 已提交
78
  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
79
  if (p && IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
80 81 82
    dbFName = p + 1;
  }

D
dapan1121 已提交
83
  SCtgDBCache *dbCache = (SCtgDBCache *)taosHashGet(pCtg->dbCache, dbFName, strlen(dbFName));
D
dapan1121 已提交
84 85 86 87 88 89
  if (NULL == dbCache) {
    *pCache = NULL;
    ctgDebug("db not in cache, dbFName:%s", dbFName);
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
90 91 92 93
  if (acquire) {
    CTG_LOCK(CTG_READ, &dbCache->dbLock);
  }

D
dapan1121 已提交
94 95 96
  if (dbCache->deleted) {
    if (acquire) {
      ctgReleaseDBCache(pCtg, dbCache);
dengyihao's avatar
dengyihao 已提交
97 98
    }

D
dapan1121 已提交
99 100 101 102 103 104
    *pCache = NULL;
    ctgDebug("db is removing from cache, dbFName:%s", dbFName);
    return TSDB_CODE_SUCCESS;
  }

  *pCache = dbCache;
dengyihao's avatar
dengyihao 已提交
105

D
dapan1121 已提交
106 107 108
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
109
int32_t ctgAcquireDBCache(SCatalog *pCtg, const char *dbFName, SCtgDBCache **pCache) {
D
dapan1121 已提交
110 111 112
  CTG_RET(ctgAcquireDBCacheImpl(pCtg, dbFName, pCache, true));
}

dengyihao's avatar
dengyihao 已提交
113
int32_t ctgGetDBCache(SCatalog *pCtg, const char *dbFName, SCtgDBCache **pCache) {
D
dapan1121 已提交
114 115 116
  CTG_RET(ctgAcquireDBCacheImpl(pCtg, dbFName, pCache, false));
}

dengyihao's avatar
dengyihao 已提交
117
void ctgReleaseVgInfoToCache(SCatalog *pCtg, SCtgDBCache *dbCache) {
D
dapan1121 已提交
118 119 120
  ctgRUnlockVgInfo(dbCache);
  ctgReleaseDBCache(pCtg, dbCache);
}
D
dapan1121 已提交
121

dengyihao's avatar
dengyihao 已提交
122
void ctgReleaseTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, SCtgTbCache *pCache) {
D
dapan1121 已提交
123 124
  if (pCache) {
    CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
125
    taosHashRelease(dbCache->tbCache, pCache);
D
dapan1121 已提交
126
  }
D
dapan1121 已提交
127

D
dapan1121 已提交
128 129
  if (dbCache) {
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
130
  }
D
dapan1121 已提交
131
}
D
dapan1121 已提交
132

dengyihao's avatar
dengyihao 已提交
133
void ctgReleaseTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, SCtgTbCache *pCache) {
D
dapan1121 已提交
134 135
  if (pCache) {
    CTG_UNLOCK(CTG_READ, &pCache->indexLock);
dengyihao's avatar
dengyihao 已提交
136
    taosHashRelease(dbCache->tbCache, pCache);
D
dapan1121 已提交
137 138 139 140 141 142 143
  }

  if (dbCache) {
    ctgReleaseDBCache(pCtg, dbCache);
  }
}

dengyihao's avatar
dengyihao 已提交
144
int32_t ctgAcquireVgInfoFromCache(SCatalog *pCtg, const char *dbFName, SCtgDBCache **pCache) {
D
dapan1121 已提交
145
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
146
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
dengyihao's avatar
dengyihao 已提交
147
  if (NULL == dbCache) {
D
dapan1121 已提交
148 149 150 151 152
    ctgDebug("db %s not in cache", dbFName);
    goto _return;
  }

  bool inCache = false;
D
dapan1121 已提交
153
  ctgRLockVgInfo(pCtg, dbCache, &inCache);
D
dapan1121 已提交
154 155 156 157 158 159 160
  if (!inCache) {
    ctgDebug("vgInfo of db %s not in cache", dbFName);
    goto _return;
  }

  *pCache = dbCache;

D
dapan1121 已提交
161
  CTG_CACHE_STAT_INC(numOfVgHit, 1);
D
dapan1121 已提交
162 163

  ctgDebug("Got db vgInfo from cache, dbFName:%s", dbFName);
dengyihao's avatar
dengyihao 已提交
164

D
dapan1121 已提交
165 166 167 168 169 170 171 172 173 174
  return TSDB_CODE_SUCCESS;

_return:

  if (dbCache) {
    ctgReleaseDBCache(pCtg, dbCache);
  }

  *pCache = NULL;

D
dapan1121 已提交
175
  CTG_CACHE_STAT_INC(numOfVgMiss, 1);
dengyihao's avatar
dengyihao 已提交
176

D
dapan1121 已提交
177 178 179
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
180
int32_t ctgAcquireTbMetaFromCache(SCatalog *pCtg, char *dbFName, char *tbName, SCtgDBCache **pDb, SCtgTbCache **pTb) {
D
dapan1121 已提交
181
  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
182
  SCtgTbCache *pCache = NULL;
D
dapan1121 已提交
183 184 185 186 187
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
    ctgDebug("db %s not in cache", dbFName);
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
188

D
dapan1121 已提交
189
  pCache = taosHashAcquire(dbCache->tbCache, tbName, strlen(tbName));
D
dapan1121 已提交
190 191 192
  if (NULL == pCache) {
    ctgDebug("tb %s not in cache, dbFName:%s", tbName, dbFName);
    goto _return;
D
dapan1121 已提交
193 194
  }

D
dapan1121 已提交
195 196 197 198 199 200 201 202 203 204
  CTG_LOCK(CTG_READ, &pCache->metaLock);
  if (NULL == pCache->pMeta) {
    ctgDebug("tb %s meta not in cache, dbFName:%s", tbName, dbFName);
    goto _return;
  }

  *pDb = dbCache;
  *pTb = pCache;

  ctgDebug("tb %s meta got in cache, dbFName:%s", tbName, dbFName);
dengyihao's avatar
dengyihao 已提交
205

D
dapan1121 已提交
206
  CTG_CACHE_STAT_INC(numOfMetaHit, 1);
D
dapan1121 已提交
207 208 209 210 211 212 213

  return TSDB_CODE_SUCCESS;

_return:

  ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);

D
dapan1121 已提交
214
  CTG_CACHE_STAT_INC(numOfMetaMiss, 1);
dengyihao's avatar
dengyihao 已提交
215

D
dapan1121 已提交
216 217 218
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
219 220 221
int32_t ctgAcquireStbMetaFromCache(SCatalog *pCtg, char *dbFName, uint64_t suid, SCtgDBCache **pDb, SCtgTbCache **pTb) {
  SCtgDBCache *dbCache = NULL;
  SCtgTbCache *pCache = NULL;
D
dapan1121 已提交
222 223 224 225 226
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
    ctgDebug("db %s not in cache", dbFName);
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
227 228

  char *stName = taosHashAcquire(dbCache->stbCache, &suid, sizeof(suid));
D
dapan1121 已提交
229
  if (NULL == stName) {
D
dapan1121 已提交
230
    ctgDebug("stb 0x%" PRIx64 " not in cache, dbFName:%s", suid, dbFName);
D
dapan1121 已提交
231 232 233 234 235
    goto _return;
  }

  pCache = taosHashAcquire(dbCache->tbCache, stName, strlen(stName));
  if (NULL == pCache) {
D
dapan1121 已提交
236
    ctgDebug("stb 0x%" PRIx64 " name %s not in cache, dbFName:%s", suid, stName, dbFName);
D
dapan1121 已提交
237 238 239 240 241 242
    taosHashRelease(dbCache->stbCache, stName);
    goto _return;
  }

  CTG_LOCK(CTG_READ, &pCache->metaLock);
  if (NULL == pCache->pMeta) {
D
dapan1121 已提交
243
    ctgDebug("stb 0x%" PRIx64 " meta not in cache, dbFName:%s", suid, dbFName);
D
dapan1121 已提交
244 245 246 247 248 249
    goto _return;
  }

  *pDb = dbCache;
  *pTb = pCache;

D
dapan1121 已提交
250
  ctgDebug("stb 0x%" PRIx64 " meta got in cache, dbFName:%s", suid, dbFName);
dengyihao's avatar
dengyihao 已提交
251

D
dapan1121 已提交
252
  CTG_CACHE_STAT_INC(numOfMetaHit, 1);
D
dapan1121 已提交
253 254 255 256 257 258 259

  return TSDB_CODE_SUCCESS;

_return:

  ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);

D
dapan1121 已提交
260
  CTG_CACHE_STAT_INC(numOfMetaMiss, 1);
D
dapan1121 已提交
261 262 263

  *pDb = NULL;
  *pTb = NULL;
dengyihao's avatar
dengyihao 已提交
264

D
dapan1121 已提交
265 266 267
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
268
int32_t ctgAcquireTbIndexFromCache(SCatalog *pCtg, char *dbFName, char *tbName, SCtgDBCache **pDb, SCtgTbCache **pTb) {
D
dapan1121 已提交
269
  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
270
  SCtgTbCache *pCache = NULL;
D
dapan1121 已提交
271 272
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
D
dapan1121 已提交
273 274 275
    ctgDebug("db %s not in cache", dbFName);
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
276

D
dapan1121 已提交
277
  int32_t sz = 0;
D
dapan1121 已提交
278
  pCache = taosHashAcquire(dbCache->tbCache, tbName, strlen(tbName));
D
dapan1121 已提交
279 280 281 282 283 284 285 286 287
  if (NULL == pCache) {
    ctgDebug("tb %s not in cache, dbFName:%s", tbName, dbFName);
    goto _return;
  }

  CTG_LOCK(CTG_READ, &pCache->indexLock);
  if (NULL == pCache->pIndex) {
    ctgDebug("tb %s index not in cache, dbFName:%s", tbName, dbFName);
    goto _return;
D
dapan1121 已提交
288 289
  }

D
dapan1121 已提交
290 291 292 293
  *pDb = dbCache;
  *pTb = pCache;

  ctgDebug("tb %s index got in cache, dbFName:%s", tbName, dbFName);
dengyihao's avatar
dengyihao 已提交
294

D
dapan1121 已提交
295
  CTG_CACHE_STAT_INC(numOfIndexHit, 1);
D
dapan1121 已提交
296 297 298 299 300 301 302

  return TSDB_CODE_SUCCESS;

_return:

  ctgReleaseTbIndexToCache(pCtg, dbCache, pCache);

D
dapan1121 已提交
303
  CTG_CACHE_STAT_INC(numOfIndexMiss, 1);
dengyihao's avatar
dengyihao 已提交
304

D
dapan1121 已提交
305 306 307
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
308
int32_t ctgTbMetaExistInCache(SCatalog *pCtg, char *dbFName, char *tbName, int32_t *exist) {
D
dapan1121 已提交
309
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
310
  SCtgTbCache *tbCache = NULL;
D
dapan1121 已提交
311 312 313
  ctgAcquireTbMetaFromCache(pCtg, dbFName, tbName, &dbCache, &tbCache);
  if (NULL == tbCache) {
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
dengyihao's avatar
dengyihao 已提交
314

D
dapan1121 已提交
315 316 317 318 319
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }

  *exist = 1;
D
dapan1121 已提交
320
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
dengyihao's avatar
dengyihao 已提交
321

D
dapan1121 已提交
322 323 324
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
325 326
int32_t ctgReadTbMetaFromCache(SCatalog *pCtg, SCtgTbMetaCtx *ctx, STableMeta **pTableMeta) {
  int32_t      code = 0;
D
dapan1121 已提交
327
  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
328
  SCtgTbCache *tbCache = NULL;
D
dapan1121 已提交
329 330 331 332 333 334 335 336 337
  *pTableMeta = NULL;

  char dbFName[TSDB_DB_FNAME_LEN] = {0};
  if (CTG_FLAG_IS_SYS_DB(ctx->flag)) {
    strcpy(dbFName, ctx->pName->dbname);
  } else {
    tNameGetFullDbName(ctx->pName, dbFName);
  }

D
dapan1121 已提交
338 339 340
  ctgAcquireTbMetaFromCache(pCtg, dbFName, ctx->pName->tname, &dbCache, &tbCache);
  if (NULL == tbCache) {
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
341 342 343
    return TSDB_CODE_SUCCESS;
  }

dengyihao's avatar
dengyihao 已提交
344
  STableMeta *tbMeta = tbCache->pMeta;
D
dapan1121 已提交
345 346 347 348
  ctx->tbInfo.inCache = true;
  ctx->tbInfo.dbId = dbCache->dbId;
  ctx->tbInfo.suid = tbMeta->suid;
  ctx->tbInfo.tbType = tbMeta->tableType;
dengyihao's avatar
dengyihao 已提交
349

D
dapan1121 已提交
350
  if (tbMeta->tableType != TSDB_CHILD_TABLE) {
D
dapan1121 已提交
351 352 353 354 355 356 357 358
    int32_t metaSize = CTG_META_SIZE(tbMeta);
    *pTableMeta = taosMemoryCalloc(1, metaSize);
    if (NULL == *pTableMeta) {
      ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
    }

    memcpy(*pTableMeta, tbMeta, metaSize);
dengyihao's avatar
dengyihao 已提交
359

D
dapan1121 已提交
360 361
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
    ctgDebug("Got tb %s meta from cache, type:%d, dbFName:%s", ctx->pName->tname, tbMeta->tableType, dbFName);
D
dapan1121 已提交
362 363
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
364 365

  // PROCESS FOR CHILD TABLE
dengyihao's avatar
dengyihao 已提交
366

D
dapan1121 已提交
367 368 369 370
  int32_t metaSize = sizeof(SCTableMeta);
  *pTableMeta = taosMemoryCalloc(1, metaSize);
  if (NULL == *pTableMeta) {
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
371 372
  }

D
dapan1121 已提交
373
  memcpy(*pTableMeta, tbMeta, metaSize);
dengyihao's avatar
dengyihao 已提交
374

D
dapan1121 已提交
375
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
dengyihao's avatar
dengyihao 已提交
376 377
  ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s", ctx->pName->tname,
           ctx->tbInfo.tbType, dbFName);
D
dapan1121 已提交
378 379 380 381 382

  ctgAcquireStbMetaFromCache(pCtg, dbFName, ctx->tbInfo.suid, &dbCache, &tbCache);
  if (NULL == tbCache) {
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
383
    ctgDebug("stb 0x%" PRIx64 " meta not in cache", ctx->tbInfo.suid);
D
dapan1121 已提交
384 385
    return TSDB_CODE_SUCCESS;
  }
dengyihao's avatar
dengyihao 已提交
386 387 388

  STableMeta *stbMeta = tbCache->pMeta;
  if (stbMeta->suid != ctx->tbInfo.suid) {
D
dapan1121 已提交
389
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
dengyihao's avatar
dengyihao 已提交
390
    ctgError("stb suid 0x%" PRIx64 " in stbCache mis-match, expected suid 0x%" PRIx64, stbMeta->suid, ctx->tbInfo.suid);
D
dapan1121 已提交
391 392 393
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

D
dapan1121 已提交
394
  metaSize = CTG_META_SIZE(stbMeta);
D
dapan1121 已提交
395
  *pTableMeta = taosMemoryRealloc(*pTableMeta, metaSize);
dengyihao's avatar
dengyihao 已提交
396
  if (NULL == *pTableMeta) {
D
dapan1121 已提交
397
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
398
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
399 400
  }

D
dapan1121 已提交
401
  memcpy(&(*pTableMeta)->sversion, &stbMeta->sversion, metaSize - sizeof(SCTableMeta));
D
dapan1121 已提交
402

D
dapan1121 已提交
403
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
404

D
dapan1121 已提交
405
  ctgDebug("Got tb %s meta from cache, dbFName:%s", ctx->pName->tname, dbFName);
dengyihao's avatar
dengyihao 已提交
406

D
dapan1121 已提交
407 408 409 410
  return TSDB_CODE_SUCCESS;

_return:

D
dapan1121 已提交
411
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
412
  taosMemoryFreeClear(*pTableMeta);
dengyihao's avatar
dengyihao 已提交
413

D
dapan1121 已提交
414 415 416
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
417 418
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType,
                              uint64_t *suid, char *stbName) {
D
dapan1121 已提交
419
  *sver = -1;
D
dapan1121 已提交
420
  *tver = -1;
D
dapan1121 已提交
421 422

  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
423
  SCtgTbCache *tbCache = NULL;
D
dapan1121 已提交
424
  char         dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
425 426
  tNameGetFullDbName(pTableName, dbFName);

D
dapan1121 已提交
427 428 429
  ctgAcquireTbMetaFromCache(pCtg, dbFName, pTableName->tname, &dbCache, &tbCache);
  if (NULL == tbCache) {
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
430 431 432
    return TSDB_CODE_SUCCESS;
  }

dengyihao's avatar
dengyihao 已提交
433
  STableMeta *tbMeta = tbCache->pMeta;
D
dapan1121 已提交
434 435
  *tbType = tbMeta->tableType;
  *suid = tbMeta->suid;
D
dapan1121 已提交
436

D
dapan1121 已提交
437
  if (*tbType != TSDB_CHILD_TABLE) {
D
dapan1121 已提交
438 439 440
    *sver = tbMeta->sversion;
    *tver = tbMeta->tversion;

dengyihao's avatar
dengyihao 已提交
441 442
    ctgDebug("Got tb %s ver from cache, dbFName:%s, tbType:%d, sver:%d, tver:%d, suid:0x%" PRIx64, pTableName->tname,
             dbFName, *tbType, *sver, *tver, *suid);
D
dapan1121 已提交
443

D
dapan1121 已提交
444
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
445 446 447
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
448
  // PROCESS FOR CHILD TABLE
dengyihao's avatar
dengyihao 已提交
449

D
dapan1121 已提交
450 451
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
  ctgDebug("Got ctb %s ver from cache, will continue to get its stb ver, dbFName:%s", pTableName->tname, dbFName);
dengyihao's avatar
dengyihao 已提交
452

D
dapan1121 已提交
453 454
  ctgAcquireStbMetaFromCache(pCtg, dbFName, *suid, &dbCache, &tbCache);
  if (NULL == tbCache) {
D
dapan1121 已提交
455
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
456
    ctgDebug("stb 0x%" PRIx64 " meta not in cache", *suid);
D
dapan1121 已提交
457 458
    return TSDB_CODE_SUCCESS;
  }
dengyihao's avatar
dengyihao 已提交
459 460

  STableMeta *stbMeta = tbCache->pMeta;
D
dapan1121 已提交
461
  if (stbMeta->suid != *suid) {
D
dapan1121 已提交
462
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
dengyihao's avatar
dengyihao 已提交
463
    ctgError("stb suid 0x%" PRIx64 " in stbCache mis-match, expected suid:0x%" PRIx64, stbMeta->suid, *suid);
D
dapan1121 已提交
464 465 466
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

D
dapan1121 已提交
467
  size_t nameLen = 0;
D
dapan1121 已提交
468
  char  *name = taosHashGetKey(tbCache, &nameLen);
D
dapan1121 已提交
469 470 471 472

  strncpy(stbName, name, nameLen);
  stbName[nameLen] = 0;

D
dapan1121 已提交
473 474
  *sver = stbMeta->sversion;
  *tver = stbMeta->tversion;
D
dapan1121 已提交
475

D
dapan1121 已提交
476
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
477

dengyihao's avatar
dengyihao 已提交
478 479
  ctgDebug("Got tb %s sver %d tver %d from cache, type:%d, dbFName:%s", pTableName->tname, *sver, *tver, *tbType,
           dbFName);
D
dapan1121 已提交
480 481 482 483

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
484
int32_t ctgReadTbTypeFromCache(SCatalog *pCtg, char *dbFName, char *tbName, int32_t *tbType) {
D
dapan1121 已提交
485
  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
486
  SCtgTbCache *tbCache = NULL;
D
dapan1121 已提交
487
  CTG_ERR_RET(ctgAcquireTbMetaFromCache(pCtg, dbFName, tbName, &dbCache, &tbCache));
D
dapan1121 已提交
488 489
  if (NULL == tbCache) {
    ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
490 491
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
492 493 494 495

  *tbType = tbCache->pMeta->tableType;
  ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);

dengyihao's avatar
dengyihao 已提交
496 497
  ctgDebug("Got tb %s tbType %d from cache, dbFName:%s", tbName, *tbType, dbFName);

D
dapan1121 已提交
498 499 500
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
501 502
int32_t ctgReadTbIndexFromCache(SCatalog *pCtg, SName *pTableName, SArray **pRes) {
  int32_t      code = 0;
D
dapan1121 已提交
503
  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
504
  SCtgTbCache *tbCache = NULL;
D
dapan1121 已提交
505 506
  char         dbFName[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
507

D
dapan1121 已提交
508
  *pRes = NULL;
D
dapan1121 已提交
509

D
dapan1121 已提交
510 511 512
  ctgAcquireTbIndexFromCache(pCtg, dbFName, pTableName->tname, &dbCache, &tbCache);
  if (NULL == tbCache) {
    ctgReleaseTbIndexToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
513 514 515
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
516
  CTG_ERR_JRET(ctgCloneTableIndex(tbCache->pIndex->pIndex, pRes));
D
dapan1121 已提交
517

D
dapan1121 已提交
518
_return:
D
dapan1121 已提交
519

D
dapan1121 已提交
520
  ctgReleaseTbIndexToCache(pCtg, dbCache, tbCache);
D
dapan1121 已提交
521

D
dapan1121 已提交
522
  CTG_RET(code);
D
dapan1121 已提交
523 524
}

dengyihao's avatar
dengyihao 已提交
525
int32_t ctgChkAuthFromCache(SCatalog *pCtg, char *user, char *dbFName, AUTH_TYPE type, bool *inCache, bool *pass) {
526 527 528 529 530 531
  char *p = strchr(dbFName, '.');
  if (p) {
    ++p;
  } else {
    p = dbFName;
  }
dengyihao's avatar
dengyihao 已提交
532

533 534 535 536 537 538 539
  if (IS_SYS_DBNAME(p)) {
    *inCache = true;
    *pass = true;
    ctgDebug("sysdb %s, pass", dbFName);
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
540 541 542 543 544 545 546 547 548
  SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, user, strlen(user));
  if (NULL == pUser) {
    ctgDebug("user not in cache, user:%s", user);
    goto _return;
  }

  *inCache = true;

  ctgDebug("Got user from cache, user:%s", user);
D
dapan1121 已提交
549
  CTG_CACHE_STAT_INC(numOfUserHit, 1);
dengyihao's avatar
dengyihao 已提交
550

D
dapan1121 已提交
551 552 553 554 555 556 557 558 559 560 561
  if (pUser->superUser) {
    *pass = true;
    return TSDB_CODE_SUCCESS;
  }

  CTG_LOCK(CTG_READ, &pUser->lock);
  if (pUser->createdDbs && taosHashGet(pUser->createdDbs, dbFName, strlen(dbFName))) {
    *pass = true;
    CTG_UNLOCK(CTG_READ, &pUser->lock);
    return TSDB_CODE_SUCCESS;
  }
dengyihao's avatar
dengyihao 已提交
562

D
dapan1121 已提交
563 564 565
  if (pUser->readDbs && taosHashGet(pUser->readDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_READ) {
    *pass = true;
  }
dengyihao's avatar
dengyihao 已提交
566

D
dapan1121 已提交
567 568 569 570 571
  if (pUser->writeDbs && taosHashGet(pUser->writeDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_WRITE) {
    *pass = true;
  }

  CTG_UNLOCK(CTG_READ, &pUser->lock);
dengyihao's avatar
dengyihao 已提交
572

D
dapan1121 已提交
573 574 575 576 577
  return TSDB_CODE_SUCCESS;

_return:

  *inCache = false;
D
dapan1121 已提交
578
  CTG_CACHE_STAT_INC(numOfUserMiss, 1);
dengyihao's avatar
dengyihao 已提交
579

D
dapan1121 已提交
580 581 582
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
583
void ctgDequeue(SCtgCacheOperation **op) {
D
dapan1121 已提交
584
  SCtgQNode *orig = gCtgMgmt.queue.head;
dengyihao's avatar
dengyihao 已提交
585

D
dapan1121 已提交
586 587 588
  SCtgQNode *node = gCtgMgmt.queue.head->next;
  gCtgMgmt.queue.head = gCtgMgmt.queue.head->next;

D
dapan1121 已提交
589
  CTG_QUEUE_DEC();
dengyihao's avatar
dengyihao 已提交
590

D
dapan1121 已提交
591 592
  taosMemoryFreeClear(orig);

D
dapan1121 已提交
593
  *op = node->op;
D
dapan1121 已提交
594 595
}

dengyihao's avatar
dengyihao 已提交
596
int32_t ctgEnqueue(SCatalog *pCtg, SCtgCacheOperation *operation) {
D
dapan1121 已提交
597 598 599
  SCtgQNode *node = taosMemoryCalloc(1, sizeof(SCtgQNode));
  if (NULL == node) {
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
D
dapan1121 已提交
600
    CTG_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
601 602
  }

dengyihao's avatar
dengyihao 已提交
603 604
  bool  syncOp = operation->syncOp;
  char *opName = gCtgCacheOperation[operation->opId].name;
D
dapan1121 已提交
605 606 607
  if (operation->syncOp) {
    tsem_init(&operation->rspSem, 0, 0);
  }
dengyihao's avatar
dengyihao 已提交
608

D
dapan1121 已提交
609
  node->op = operation;
D
dapan1121 已提交
610 611

  CTG_LOCK(CTG_WRITE, &gCtgMgmt.queue.qlock);
D
dapan1121 已提交
612
  if (gCtgMgmt.queue.stopQueue) {
613 614 615 616
    ctgFreeQNode(node);
    CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.queue.qlock);
    CTG_RET(TSDB_CODE_CTG_EXIT);
  }
D
dapan1121 已提交
617
  gCtgMgmt.queue.stopQueue = operation->stopQueue;
D
dapan1121 已提交
618 619 620 621
  gCtgMgmt.queue.tail->next = node;
  gCtgMgmt.queue.tail = node;
  CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.queue.qlock);

D
dapan1121 已提交
622 623
  ctgDebug("action [%s] added into queue", opName);

D
dapan1121 已提交
624
  CTG_QUEUE_INC();
D
dapan1121 已提交
625
  CTG_RT_STAT_INC(numOfOpEnqueue, 1);
D
dapan1121 已提交
626 627 628

  tsem_post(&gCtgMgmt.queue.reqSem);

D
dapan1121 已提交
629
  if (syncOp) {
630 631 632
    if (!operation->unLocked) {
      CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
    }
D
dapan1121 已提交
633
    tsem_wait(&operation->rspSem);
634 635 636
    if (!operation->unLocked) {
      CTG_LOCK(CTG_READ, &gCtgMgmt.lock);
    }
D
dapan1121 已提交
637
    taosMemoryFree(operation);
D
dapan1121 已提交
638 639 640 641 642
  }

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
643 644
int32_t ctgDropDbCacheEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId) {
  int32_t             code = 0;
D
dapan1121 已提交
645 646
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_DROP_DB_CACHE;
dengyihao's avatar
dengyihao 已提交
647

D
dapan1121 已提交
648
  SCtgDropDBMsg *msg = taosMemoryMalloc(sizeof(SCtgDropDBMsg));
D
dapan1121 已提交
649
  if (NULL == msg) {
D
dapan1121 已提交
650
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropDBMsg));
D
dapan1121 已提交
651
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
652 653 654
  }

  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
655
  if (p && IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
656 657 658 659 660 661 662
    dbFName = p + 1;
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;

D
dapan1121 已提交
663
  op->data = msg;
D
dapan1121 已提交
664

D
dapan1121 已提交
665
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
666 667 668 669 670

  return TSDB_CODE_SUCCESS;

_return:

D
dapan1121 已提交
671
  taosMemoryFreeClear(op->data);
D
dapan1121 已提交
672 673 674
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
675 676
int32_t ctgDropDbVgroupEnqueue(SCatalog *pCtg, const char *dbFName, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
677 678 679
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_DROP_DB_VGROUP;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
680

D
dapan1121 已提交
681 682 683
  SCtgDropDbVgroupMsg *msg = taosMemoryMalloc(sizeof(SCtgDropDbVgroupMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropDbVgroupMsg));
D
dapan1121 已提交
684
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
685 686 687
  }

  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
688
  if (p && IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
689 690 691 692 693 694
    dbFName = p + 1;
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));

D
dapan1121 已提交
695
  op->data = msg;
D
dapan1121 已提交
696

D
dapan1121 已提交
697
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
698 699 700 701 702

  return TSDB_CODE_SUCCESS;

_return:

D
dapan1121 已提交
703
  taosMemoryFreeClear(op->data);
D
dapan1121 已提交
704 705 706
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
707 708 709
int32_t ctgDropStbMetaEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid,
                              bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
710 711 712
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_DROP_STB_META;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
713

D
dapan1121 已提交
714
  SCtgDropStbMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgDropStbMetaMsg));
D
dapan1121 已提交
715
  if (NULL == msg) {
D
dapan1121 已提交
716
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropStbMetaMsg));
D
dapan1121 已提交
717
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
718 719 720 721 722 723 724 725
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  strncpy(msg->stbName, stbName, sizeof(msg->stbName));
  msg->dbId = dbId;
  msg->suid = suid;

D
dapan1121 已提交
726
  op->data = msg;
D
dapan1121 已提交
727

D
dapan1121 已提交
728
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
729 730 731 732 733

  return TSDB_CODE_SUCCESS;

_return:

D
dapan1121 已提交
734
  taosMemoryFreeClear(op->data);
D
dapan1121 已提交
735 736 737
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
738 739
int32_t ctgDropTbMetaEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
740 741 742
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_DROP_TB_META;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
743

D
dapan1121 已提交
744
  SCtgDropTblMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgDropTblMetaMsg));
D
dapan1121 已提交
745
  if (NULL == msg) {
D
dapan1121 已提交
746
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropTblMetaMsg));
D
dapan1121 已提交
747
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
748 749 750 751 752 753 754
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  strncpy(msg->tbName, tbName, sizeof(msg->tbName));
  msg->dbId = dbId;

D
dapan1121 已提交
755
  op->data = msg;
D
dapan1121 已提交
756

D
dapan1121 已提交
757
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
758 759 760 761 762

  return TSDB_CODE_SUCCESS;

_return:

D
dapan1121 已提交
763
  taosMemoryFreeClear(op->data);
D
dapan1121 已提交
764 765 766
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
767 768
int32_t ctgUpdateVgroupEnqueue(SCatalog *pCtg, const char *dbFName, int64_t dbId, SDBVgInfo *dbInfo, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
769 770 771
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_UPDATE_VGROUP;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
772

D
dapan1121 已提交
773 774 775 776
  SCtgUpdateVgMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateVgMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateVgMsg));
    ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
777
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
778 779 780
  }

  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
781
  if (p && IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
782 783 784 785 786 787 788 789
    dbFName = p + 1;
  }

  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->pCtg = pCtg;
  msg->dbId = dbId;
  msg->dbInfo = dbInfo;

D
dapan1121 已提交
790
  op->data = msg;
D
dapan1121 已提交
791

D
dapan1121 已提交
792
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
793 794 795 796 797 798

  return TSDB_CODE_SUCCESS;

_return:

  ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
799
  taosMemoryFreeClear(op->data);
D
dapan1121 已提交
800
  taosMemoryFreeClear(op);
D
dapan1121 已提交
801 802 803
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
804 805
int32_t ctgUpdateTbMetaEnqueue(SCatalog *pCtg, STableMetaOutput *output, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
806 807 808
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_UPDATE_TB_META;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
809

D
dapan1121 已提交
810
  SCtgUpdateTbMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateTbMetaMsg));
D
dapan1121 已提交
811
  if (NULL == msg) {
D
dapan1121 已提交
812
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateTbMetaMsg));
D
dapan1121 已提交
813
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
814 815 816
  }

  char *p = strchr(output->dbFName, '.');
D
dapan1121 已提交
817
  if (p && IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
818 819 820 821
    memmove(output->dbFName, p + 1, strlen(p + 1));
  }

  msg->pCtg = pCtg;
D
dapan1121 已提交
822
  msg->pMeta = output;
D
dapan1121 已提交
823

D
dapan1121 已提交
824
  op->data = msg;
D
dapan1121 已提交
825

D
dapan1121 已提交
826
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
827 828

  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
829

D
dapan1121 已提交
830 831
_return:

D
dapan1121 已提交
832 833 834 835 836
  if (output) {
    taosMemoryFree(output->tbMeta);
    taosMemoryFree(output);
  }

D
dapan1121 已提交
837
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
838

D
dapan1121 已提交
839 840 841
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
842 843
int32_t ctgUpdateVgEpsetEnqueue(SCatalog *pCtg, char *dbFName, int32_t vgId, SEpSet *pEpSet) {
  int32_t             code = 0;
D
dapan1121 已提交
844 845
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_UPDATE_VG_EPSET;
dengyihao's avatar
dengyihao 已提交
846

D
dapan1121 已提交
847 848 849
  SCtgUpdateEpsetMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateEpsetMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateEpsetMsg));
D
dapan1121 已提交
850
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
851 852 853 854 855 856 857
  }

  msg->pCtg = pCtg;
  strcpy(msg->dbFName, dbFName);
  msg->vgId = vgId;
  msg->epSet = *pEpSet;

D
dapan1121 已提交
858
  op->data = msg;
D
dapan1121 已提交
859

D
dapan1121 已提交
860
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
861 862

  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
863

D
dapan1121 已提交
864 865 866
_return:

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
867

D
dapan1121 已提交
868 869 870
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
871 872
int32_t ctgUpdateUserEnqueue(SCatalog *pCtg, SGetUserAuthRsp *pAuth, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
873 874 875
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_UPDATE_USER;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
876

D
dapan1121 已提交
877 878 879
  SCtgUpdateUserMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateUserMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateUserMsg));
D
dapan1121 已提交
880
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
881 882 883 884 885
  }

  msg->pCtg = pCtg;
  msg->userAuth = *pAuth;

D
dapan1121 已提交
886
  op->data = msg;
D
dapan1121 已提交
887

D
dapan1121 已提交
888
  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
dengyihao's avatar
dengyihao 已提交
889

D
dapan1121 已提交
890
  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
891

D
dapan1121 已提交
892 893 894 895
_return:

  tFreeSGetUserAuthRsp(pAuth);
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
896

D
dapan1121 已提交
897 898 899
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
900 901
int32_t ctgUpdateTbIndexEnqueue(SCatalog *pCtg, STableIndex **pIndex, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
902 903 904
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_UPDATE_TB_INDEX;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
905

D
dapan1121 已提交
906 907 908
  SCtgUpdateTbIndexMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateTbIndexMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateTbIndexMsg));
D
dapan1121 已提交
909
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
910 911 912
  }

  msg->pCtg = pCtg;
D
dapan1121 已提交
913
  msg->pIndex = *pIndex;
D
dapan1121 已提交
914 915 916 917

  op->data = msg;

  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
D
dapan1121 已提交
918 919

  *pIndex = NULL;
D
dapan1121 已提交
920
  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
921

D
dapan1121 已提交
922 923
_return:

D
dapan1121 已提交
924
  taosArrayDestroyEx((*pIndex)->pIndex, tFreeSTableIndexInfo);
D
dapan1121 已提交
925
  taosMemoryFreeClear(*pIndex);
D
dapan1121 已提交
926
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
927

D
dapan1121 已提交
928 929 930
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
931 932
int32_t ctgDropTbIndexEnqueue(SCatalog *pCtg, SName *pName, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
933 934 935
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_DROP_TB_INDEX;
  op->syncOp = syncOp;
dengyihao's avatar
dengyihao 已提交
936

D
dapan1121 已提交
937 938 939
  SCtgDropTbIndexMsg *msg = taosMemoryMalloc(sizeof(SCtgDropTbIndexMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropTbIndexMsg));
D
dapan1121 已提交
940
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
941 942 943 944 945 946 947 948 949
  }

  msg->pCtg = pCtg;
  tNameGetFullDbName(pName, msg->dbFName);
  strcpy(msg->tbName, pName->tname);

  op->data = msg;

  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
dengyihao's avatar
dengyihao 已提交
950

D
dapan1121 已提交
951
  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
952

D
dapan1121 已提交
953 954 955
_return:

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
956

D
dapan1121 已提交
957 958 959
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
960 961
int32_t ctgClearCacheEnqueue(SCatalog *pCtg, bool freeCtg, bool stopQueue, bool syncOp) {
  int32_t             code = 0;
D
dapan1121 已提交
962 963 964
  SCtgCacheOperation *op = taosMemoryCalloc(1, sizeof(SCtgCacheOperation));
  op->opId = CTG_OP_CLEAR_CACHE;
  op->syncOp = syncOp;
D
dapan1121 已提交
965
  op->stopQueue = stopQueue;
966
  op->unLocked = true;
dengyihao's avatar
dengyihao 已提交
967

D
dapan1121 已提交
968 969 970
  SCtgClearCacheMsg *msg = taosMemoryMalloc(sizeof(SCtgClearCacheMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgClearCacheMsg));
D
dapan1121 已提交
971
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
972 973 974
  }

  msg->pCtg = pCtg;
D
dapan1121 已提交
975
  msg->freeCtg = freeCtg;
D
dapan1121 已提交
976 977 978
  op->data = msg;

  CTG_ERR_JRET(ctgEnqueue(pCtg, op));
dengyihao's avatar
dengyihao 已提交
979

D
dapan1121 已提交
980
  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
981

D
dapan1121 已提交
982 983 984
_return:

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
985

D
dapan1121 已提交
986 987 988
  CTG_RET(code);
}

D
dapan1121 已提交
989 990 991 992 993 994
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
  mgmt->slotRIdx = 0;
  mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
  mgmt->type = type;

  size_t msgSize = sizeof(SCtgRentSlot) * mgmt->slotNum;
dengyihao's avatar
dengyihao 已提交
995

D
dapan1121 已提交
996 997 998
  mgmt->slots = taosMemoryCalloc(1, msgSize);
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
D
dapan1121 已提交
999
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1000 1001 1002
  }

  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
dengyihao's avatar
dengyihao 已提交
1003

D
dapan1121 已提交
1004 1005 1006 1007 1008 1009 1010
  return TSDB_CODE_SUCCESS;
}

int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size) {
  int16_t widx = abs((int)(id % mgmt->slotNum));

  SCtgRentSlot *slot = &mgmt->slots[widx];
dengyihao's avatar
dengyihao 已提交
1011 1012
  int32_t       code = 0;

D
dapan1121 已提交
1013 1014 1015 1016
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
    slot->meta = taosArrayInit(CTG_DEFAULT_RENT_SLOT_SIZE, size);
    if (NULL == slot->meta) {
dengyihao's avatar
dengyihao 已提交
1017 1018
      qError("taosArrayInit %d failed, id:0x%" PRIx64 ", slot idx:%d, type:%d", CTG_DEFAULT_RENT_SLOT_SIZE, id, widx,
             mgmt->type);
D
dapan1121 已提交
1019
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1020 1021 1022 1023
    }
  }

  if (NULL == taosArrayPush(slot->meta, meta)) {
dengyihao's avatar
dengyihao 已提交
1024
    qError("taosArrayPush meta to rent failed, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1025
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1026 1027 1028 1029
  }

  slot->needSort = true;

dengyihao's avatar
dengyihao 已提交
1030
  qDebug("add meta to rent, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1031 1032 1033 1034 1035 1036 1037

_return:

  CTG_UNLOCK(CTG_WRITE, &slot->lock);
  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
1038 1039
int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t sortCompare,
                          __compar_fn_t searchCompare) {
D
dapan1121 已提交
1040 1041 1042
  int16_t widx = abs((int)(id % mgmt->slotNum));

  SCtgRentSlot *slot = &mgmt->slots[widx];
dengyihao's avatar
dengyihao 已提交
1043
  int32_t       code = 0;
D
dapan1121 已提交
1044 1045 1046

  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
dengyihao's avatar
dengyihao 已提交
1047
    qDebug("empty meta slot, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1048 1049 1050 1051
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  if (slot->needSort) {
dengyihao's avatar
dengyihao 已提交
1052 1053
    qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type,
           (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1054 1055 1056 1057 1058 1059 1060
    taosArraySort(slot->meta, sortCompare);
    slot->needSort = false;
    qDebug("meta slot sorted, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
  }

  void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ);
  if (NULL == orig) {
dengyihao's avatar
dengyihao 已提交
1061 1062
    qDebug("meta not found in slot, id:0x%" PRIx64 ", slot idx:%d, type:%d, size:%d", id, widx, mgmt->type,
           (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1063 1064 1065 1066 1067
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  memcpy(orig, meta, size);

dengyihao's avatar
dengyihao 已提交
1068
  qDebug("meta in rent updated, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1069 1070 1071 1072 1073 1074

_return:

  CTG_UNLOCK(CTG_WRITE, &slot->lock);

  if (code) {
dengyihao's avatar
dengyihao 已提交
1075 1076
    qDebug("meta in rent update failed, will try to add it, code:%x, id:0x%" PRIx64 ", slot idx:%d, type:%d", code, id,
           widx, mgmt->type);
D
dapan1121 已提交
1077 1078 1079 1080 1081 1082 1083 1084 1085 1086
    CTG_RET(ctgMetaRentAdd(mgmt, meta, id, size));
  }

  CTG_RET(code);
}

int32_t ctgMetaRentRemove(SCtgRentMgmt *mgmt, int64_t id, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
  int16_t widx = abs((int)(id % mgmt->slotNum));

  SCtgRentSlot *slot = &mgmt->slots[widx];
dengyihao's avatar
dengyihao 已提交
1087 1088
  int32_t       code = 0;

D
dapan1121 已提交
1089 1090
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
dengyihao's avatar
dengyihao 已提交
1091
    qError("empty meta slot, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  if (slot->needSort) {
    taosArraySort(slot->meta, sortCompare);
    slot->needSort = false;
    qDebug("meta slot sorted, slot idx:%d, type:%d", widx, mgmt->type);
  }

  int32_t idx = taosArraySearchIdx(slot->meta, &id, searchCompare, TD_EQ);
  if (idx < 0) {
dengyihao's avatar
dengyihao 已提交
1103
    qError("meta not found in slot, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1104 1105 1106 1107 1108
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  taosArrayRemove(slot->meta, idx);

dengyihao's avatar
dengyihao 已提交
1109
  qDebug("meta in rent removed, id:0x%" PRIx64 ", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125

_return:

  CTG_UNLOCK(CTG_WRITE, &slot->lock);

  CTG_RET(code);
}

int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
  int16_t ridx = atomic_add_fetch_16(&mgmt->slotRIdx, 1);
  if (ridx >= mgmt->slotNum) {
    ridx %= mgmt->slotNum;
    atomic_store_16(&mgmt->slotRIdx, ridx);
  }

  SCtgRentSlot *slot = &mgmt->slots[ridx];
dengyihao's avatar
dengyihao 已提交
1126 1127
  int32_t       code = 0;

D
dapan1121 已提交
1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145
  CTG_LOCK(CTG_READ, &slot->lock);
  if (NULL == slot->meta) {
    qDebug("empty meta in slot:%d, type:%d", ridx, mgmt->type);
    *num = 0;
    goto _return;
  }

  size_t metaNum = taosArrayGetSize(slot->meta);
  if (metaNum <= 0) {
    qDebug("no meta in slot:%d, type:%d", ridx, mgmt->type);
    *num = 0;
    goto _return;
  }

  size_t msize = metaNum * size;
  *res = taosMemoryMalloc(msize);
  if (NULL == *res) {
    qError("malloc %d failed", (int32_t)msize);
D
dapan1121 已提交
1146
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192
  }

  void *meta = taosArrayGet(slot->meta, 0);

  memcpy(*res, meta, msize);

  *num = (uint32_t)metaNum;

  qDebug("Got %d meta from rent, type:%d", (int32_t)metaNum, mgmt->type);

_return:

  CTG_UNLOCK(CTG_READ, &slot->lock);

  CTG_RET(code);
}

int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
  while (true) {
    int64_t msec = taosGetTimestampMs();
    int64_t lsec = atomic_load_64(&mgmt->lastReadMsec);
    if ((msec - lsec) < CTG_RENT_SLOT_SECOND * 1000) {
      *res = NULL;
      *num = 0;
      qDebug("too short time period to get expired meta, type:%d", mgmt->type);
      return TSDB_CODE_SUCCESS;
    }

    if (lsec != atomic_val_compare_exchange_64(&mgmt->lastReadMsec, lsec, msec)) {
      continue;
    }

    break;
  }

  CTG_ERR_RET(ctgMetaRentGetImpl(mgmt, res, num, size));

  return TSDB_CODE_SUCCESS;
}

int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
  int32_t code = 0;

  SCtgDBCache newDBCache = {0};
  newDBCache.dbId = dbId;

dengyihao's avatar
dengyihao 已提交
1193 1194
  newDBCache.tbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY),
                                    true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1195
  if (NULL == newDBCache.tbCache) {
D
dapan1121 已提交
1196
    ctgError("taosHashInit %d metaCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1197
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1198 1199
  }

dengyihao's avatar
dengyihao 已提交
1200 1201
  newDBCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT),
                                     true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1202
  if (NULL == newDBCache.stbCache) {
D
dapan1121 已提交
1203
    ctgError("taosHashInit %d stbCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1204
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1205 1206 1207 1208 1209 1210 1211 1212
  }

  code = taosHashPut(pCtg->dbCache, dbFName, strlen(dbFName), &newDBCache, sizeof(SCtgDBCache));
  if (code) {
    if (HASH_NODE_EXIST(code)) {
      ctgDebug("db already in cache, dbFName:%s", dbFName);
      goto _return;
    }
dengyihao's avatar
dengyihao 已提交
1213

D
dapan1121 已提交
1214
    ctgError("taosHashPut db to cache failed, dbFName:%s", dbFName);
D
dapan1121 已提交
1215
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1216 1217
  }

D
dapan1121 已提交
1218
  CTG_CACHE_STAT_INC(numOfDb, 1);
dengyihao's avatar
dengyihao 已提交
1219

D
dapan1121 已提交
1220 1221 1222
  SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));

dengyihao's avatar
dengyihao 已提交
1223
  ctgDebug("db added to cache, dbFName:%s, dbId:0x%" PRIx64, dbFName, dbId);
D
dapan1121 已提交
1224 1225 1226

  CTG_ERR_RET(ctgMetaRentAdd(&pCtg->dbRent, &vgVersion, dbId, sizeof(SDbVgVersion)));

dengyihao's avatar
dengyihao 已提交
1227
  ctgDebug("db added to rent, dbFName:%s, vgVersion:%d, dbId:0x%" PRIx64, dbFName, vgVersion.vgVersion, dbId);
D
dapan1121 已提交
1228 1229 1230 1231 1232 1233 1234 1235 1236 1237

  return TSDB_CODE_SUCCESS;

_return:

  ctgFreeDbCache(&newDBCache);

  CTG_RET(code);
}

dengyihao's avatar
dengyihao 已提交
1238
void ctgRemoveStbRent(SCatalog *pCtg, SCtgDBCache *dbCache) {
D
dapan1121 已提交
1239 1240 1241
  if (NULL == dbCache->stbCache) {
    return;
  }
dengyihao's avatar
dengyihao 已提交
1242

D
dapan1121 已提交
1243 1244 1245 1246
  void *pIter = taosHashIterate(dbCache->stbCache, NULL);
  while (pIter) {
    uint64_t *suid = NULL;
    suid = taosHashGetKey(pIter, NULL);
D
dapan1121 已提交
1247

dengyihao's avatar
dengyihao 已提交
1248 1249 1250
    if (TSDB_CODE_SUCCESS ==
        ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
      ctgDebug("stb removed from rent, suid:0x%" PRIx64, *suid);
D
dapan1121 已提交
1251
    }
dengyihao's avatar
dengyihao 已提交
1252

D
dapan1121 已提交
1253
    pIter = taosHashIterate(dbCache->stbCache, pIter);
D
dapan1121 已提交
1254 1255 1256
  }
}

dengyihao's avatar
dengyihao 已提交
1257
int32_t ctgRemoveDBFromCache(SCatalog *pCtg, SCtgDBCache *dbCache, const char *dbFName) {
D
dapan1121 已提交
1258
  uint64_t dbId = dbCache->dbId;
dengyihao's avatar
dengyihao 已提交
1259 1260

  ctgInfo("start to remove db from cache, dbFName:%s, dbId:0x%" PRIx64, dbFName, dbCache->dbId);
D
dapan1121 已提交
1261

D
dapan1121 已提交
1262
  CTG_LOCK(CTG_WRITE, &dbCache->dbLock);
D
dapan1121 已提交
1263

D
dapan1121 已提交
1264
  atomic_store_8(&dbCache->deleted, 1);
D
dapan1121 已提交
1265
  ctgRemoveStbRent(pCtg, dbCache);
D
dapan1121 已提交
1266 1267
  ctgFreeDbCache(dbCache);

D
dapan1121 已提交
1268 1269 1270
  CTG_UNLOCK(CTG_WRITE, &dbCache->dbLock);

  CTG_ERR_RET(ctgMetaRentRemove(&pCtg->dbRent, dbId, ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
dengyihao's avatar
dengyihao 已提交
1271
  ctgDebug("db removed from rent, dbFName:%s, dbId:0x%" PRIx64, dbFName, dbId);
D
dapan1121 已提交
1272 1273 1274 1275 1276 1277

  if (taosHashRemove(pCtg->dbCache, dbFName, strlen(dbFName))) {
    ctgInfo("taosHashRemove from dbCache failed, may be removed, dbFName:%s", dbFName);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
  }

D
dapan1121 已提交
1278
  CTG_CACHE_STAT_DEC(numOfDb, 1);
dengyihao's avatar
dengyihao 已提交
1279 1280
  ctgInfo("db removed from cache, dbFName:%s, dbId:0x%" PRIx64, dbFName, dbId);

D
dapan1121 已提交
1281 1282 1283
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
1284 1285
int32_t ctgGetAddDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId, SCtgDBCache **pCache) {
  int32_t      code = 0;
D
dapan1121 已提交
1286 1287
  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(pCtg, dbFName, &dbCache);
dengyihao's avatar
dengyihao 已提交
1288

D
dapan1121 已提交
1289
  if (dbCache) {
dengyihao's avatar
dengyihao 已提交
1290
    // TODO OPEN IT
D
dapan1121 已提交
1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306
#if 0    
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
#else
    if (0 == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }

    if (dbId && (dbCache->dbId == 0)) {
      dbCache->dbId = dbId;
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
dengyihao's avatar
dengyihao 已提交
1307

D
dapan1121 已提交
1308 1309 1310 1311 1312 1313 1314
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
#endif
    CTG_ERR_RET(ctgRemoveDBFromCache(pCtg, dbCache, dbFName));
  }
dengyihao's avatar
dengyihao 已提交
1315

D
dapan1121 已提交
1316 1317 1318 1319 1320 1321 1322 1323 1324
  CTG_ERR_RET(ctgAddNewDBCache(pCtg, dbFName, dbId));

  ctgGetDBCache(pCtg, dbFName, &dbCache);

  *pCache = dbCache;

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
1325 1326
int32_t ctgUpdateRentStbVersion(SCatalog *pCtg, char *dbFName, char *tbName, uint64_t dbId, uint64_t suid,
                                SCtgTbCache *pCache) {
D
dapan1121 已提交
1327 1328 1329 1330 1331 1332 1333 1334 1335
  SSTableVersion metaRent = {.dbId = dbId, .suid = suid};
  if (pCache->pMeta) {
    metaRent.sversion = pCache->pMeta->sversion;
    metaRent.tversion = pCache->pMeta->tversion;
  }

  if (pCache->pIndex) {
    metaRent.smaVer = pCache->pIndex->version;
  }
dengyihao's avatar
dengyihao 已提交
1336

D
dapan1121 已提交
1337 1338 1339
  strcpy(metaRent.dbFName, dbFName);
  strcpy(metaRent.stbName, tbName);

dengyihao's avatar
dengyihao 已提交
1340 1341
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableVersion),
                                ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
D
dapan1121 已提交
1342

dengyihao's avatar
dengyihao 已提交
1343 1344
  ctgDebug("db %s,0x%" PRIx64 " stb %s,0x%" PRIx64 " sver %d tver %d smaVer %d updated to stbRent", dbFName, dbId,
           tbName, suid, metaRent.sversion, metaRent.tversion, metaRent.smaVer);
D
dapan1121 已提交
1345

dengyihao's avatar
dengyihao 已提交
1346 1347
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1348

dengyihao's avatar
dengyihao 已提交
1349 1350
int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, uint64_t dbId, char *tbName,
                              STableMeta *meta, int32_t metaSize) {
D
dapan1121 已提交
1351 1352
  if (NULL == dbCache->tbCache || NULL == dbCache->stbCache) {
    taosMemoryFree(meta);
dengyihao's avatar
dengyihao 已提交
1353
    ctgError("db is dropping, dbId:0x%" PRIx64, dbCache->dbId);
D
dapan1121 已提交
1354 1355 1356
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
  }

dengyihao's avatar
dengyihao 已提交
1357 1358 1359 1360 1361 1362
  bool         isStb = meta->tableType == TSDB_SUPER_TABLE;
  SCtgTbCache *pCache = taosHashGet(dbCache->tbCache, tbName, strlen(tbName));
  STableMeta  *orig = (pCache ? pCache->pMeta : NULL);
  int8_t       origType = 0;
  uint64_t     origSuid = 0;

D
dapan1121 已提交
1363 1364 1365
  if (orig) {
    origType = orig->tableType;

dengyihao's avatar
dengyihao 已提交
1366 1367
    if (origType == meta->tableType && orig->uid == meta->uid &&
        (origType == TSDB_CHILD_TABLE || (orig->sversion >= meta->sversion && orig->tversion >= meta->tversion))) {
D
dapan1121 已提交
1368 1369
      taosMemoryFree(meta);
      ctgDebug("ignore table %s meta update", tbName);
D
dapan1121 已提交
1370 1371
      return TSDB_CODE_SUCCESS;
    }
dengyihao's avatar
dengyihao 已提交
1372

D
dapan1121 已提交
1373
    if (origType == TSDB_SUPER_TABLE) {
D
dapan1121 已提交
1374
      if (taosHashRemove(dbCache->stbCache, &orig->suid, sizeof(orig->suid))) {
dengyihao's avatar
dengyihao 已提交
1375
        ctgError("stb not exist in stbCache, dbFName:%s, stb:%s, suid:0x%" PRIx64, dbFName, tbName, orig->suid);
D
dapan1121 已提交
1376
      } else {
D
dapan1121 已提交
1377
        CTG_CACHE_STAT_DEC(numOfStb, 1);
dengyihao's avatar
dengyihao 已提交
1378
        ctgDebug("stb removed from stbCache, dbFName:%s, stb:%s, suid:0x%" PRIx64, dbFName, tbName, orig->suid);
D
dapan1121 已提交
1379
      }
dengyihao's avatar
dengyihao 已提交
1380

D
dapan1121 已提交
1381 1382 1383 1384
      origSuid = orig->suid;
    }
  }

D
dapan1121 已提交
1385 1386 1387 1388 1389 1390
  if (NULL == pCache) {
    SCtgTbCache cache = {0};
    cache.pMeta = meta;
    if (taosHashPut(dbCache->tbCache, tbName, strlen(tbName), &cache, sizeof(SCtgTbCache)) != 0) {
      taosMemoryFree(meta);
      ctgError("taosHashPut new tbCache failed, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1391
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1392
    }
dengyihao's avatar
dengyihao 已提交
1393

D
dapan1121 已提交
1394 1395 1396 1397
    pCache = taosHashGet(dbCache->tbCache, tbName, strlen(tbName));
  } else {
    taosMemoryFree(pCache->pMeta);
    pCache->pMeta = meta;
D
dapan1121 已提交
1398 1399 1400
  }

  if (NULL == orig) {
D
dapan1121 已提交
1401
    CTG_CACHE_STAT_INC(numOfTbl, 1);
D
dapan1121 已提交
1402 1403 1404 1405 1406 1407 1408 1409 1410
  }

  ctgDebug("tbmeta updated to cache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
  ctgdShowTableMeta(pCtg, tbName, meta);

  if (!isStb) {
    return TSDB_CODE_SUCCESS;
  }

dengyihao's avatar
dengyihao 已提交
1411 1412 1413
  if (origSuid != meta->suid &&
      taosHashPut(dbCache->stbCache, &meta->suid, sizeof(meta->suid), tbName, strlen(tbName) + 1) != 0) {
    ctgError("taosHashPut to stable cache failed, suid:0x%" PRIx64, meta->suid);
D
dapan1121 已提交
1414
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1415 1416
  }

D
dapan1121 已提交
1417
  CTG_CACHE_STAT_INC(numOfStb, 1);
D
dapan1121 已提交
1418

dengyihao's avatar
dengyihao 已提交
1419 1420
  ctgDebug("stb 0x%" PRIx64 " updated to cache, dbFName:%s, tbName:%s, tbType:%d", meta->suid, dbFName, tbName,
           meta->tableType);
D
dapan1121 已提交
1421

D
dapan1121 已提交
1422
  CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbId, meta->suid, pCache));
dengyihao's avatar
dengyihao 已提交
1423

D
dapan1121 已提交
1424 1425 1426
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
1427
int32_t ctgWriteTbIndexToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, char *tbName, STableIndex **index) {
D
dapan1121 已提交
1428
  if (NULL == dbCache->tbCache) {
D
dapan1121 已提交
1429
    ctgFreeSTableIndex(*index);
D
dapan1121 已提交
1430
    taosMemoryFreeClear(*index);
dengyihao's avatar
dengyihao 已提交
1431
    ctgError("db is dropping, dbId:0x%" PRIx64, dbCache->dbId);
D
dapan1121 已提交
1432 1433 1434
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
  }

dengyihao's avatar
dengyihao 已提交
1435 1436 1437
  STableIndex *pIndex = *index;
  uint64_t     suid = pIndex->suid;
  SCtgTbCache *pCache = taosHashGet(dbCache->tbCache, tbName, strlen(tbName));
D
dapan1121 已提交
1438 1439 1440
  if (NULL == pCache) {
    SCtgTbCache cache = {0};
    cache.pIndex = pIndex;
dengyihao's avatar
dengyihao 已提交
1441

D
dapan1121 已提交
1442 1443
    if (taosHashPut(dbCache->tbCache, tbName, strlen(tbName), &cache, sizeof(cache)) != 0) {
      ctgFreeSTableIndex(*index);
D
dapan1121 已提交
1444 1445
      taosMemoryFreeClear(*index);
      ctgError("taosHashPut new tbCache failed, tbName:%s", tbName);
D
dapan1121 已提交
1446
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1447 1448 1449
    }

    *index = NULL;
dengyihao's avatar
dengyihao 已提交
1450 1451
    ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version,
             (int32_t)taosArrayGetSize(pIndex->pIndex));
D
dapan1121 已提交
1452

D
dapan1121 已提交
1453
    if (suid) {
D
dapan1121 已提交
1454
      CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, pIndex->suid, &cache));
D
dapan1121 已提交
1455
    }
dengyihao's avatar
dengyihao 已提交
1456

D
dapan1121 已提交
1457 1458 1459 1460
    return TSDB_CODE_SUCCESS;
  }

  if (pCache->pIndex) {
D
dapan1121 已提交
1461 1462 1463
    if (0 == suid) {
      suid = pCache->pIndex->suid;
    }
D
dapan1121 已提交
1464 1465 1466 1467 1468 1469 1470
    taosArrayDestroyEx(pCache->pIndex->pIndex, tFreeSTableIndexInfo);
    taosMemoryFreeClear(pCache->pIndex);
  }

  pCache->pIndex = pIndex;
  *index = NULL;

dengyihao's avatar
dengyihao 已提交
1471 1472
  ctgDebug("table %s index updated to cache, ver:%d, num:%d", tbName, pIndex->version,
           (int32_t)taosArrayGetSize(pIndex->pIndex));
D
dapan1121 已提交
1473

D
dapan1121 已提交
1474 1475 1476
  if (suid) {
    CTG_ERR_RET(ctgUpdateRentStbVersion(pCtg, dbFName, tbName, dbCache->dbId, suid, pCache));
  }
dengyihao's avatar
dengyihao 已提交
1477

D
dapan1121 已提交
1478 1479 1480
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
1481 1482 1483 1484
int32_t ctgUpdateTbMetaToCache(SCatalog *pCtg, STableMetaOutput *pOut, bool syncReq) {
  STableMetaOutput *pOutput = NULL;
  int32_t           code = 0;

D
dapan1121 已提交
1485
  CTG_ERR_RET(ctgCloneMetaOutput(pOut, &pOutput));
D
dapan1121 已提交
1486
  CTG_ERR_JRET(ctgUpdateTbMetaEnqueue(pCtg, pOutput, syncReq));
D
dapan1121 已提交
1487 1488

  return TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
1489

D
dapan1121 已提交
1490 1491 1492 1493 1494 1495
_return:

  ctgFreeSTableMetaOutput(pOutput);
  CTG_RET(code);
}

D
dapan1121 已提交
1496
void ctgClearAllInstance(void) {
dengyihao's avatar
dengyihao 已提交
1497
  SCatalog *pCtg = NULL;
1498

dengyihao's avatar
dengyihao 已提交
1499
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
1500
  while (pIter) {
dengyihao's avatar
dengyihao 已提交
1501
    pCtg = *(SCatalog **)pIter;
1502 1503

    if (pCtg) {
D
dapan1121 已提交
1504 1505 1506 1507 1508 1509 1510 1511
      ctgClearHandle(pCtg);
    }

    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
  }
}

void ctgFreeAllInstance(void) {
dengyihao's avatar
dengyihao 已提交
1512
  SCatalog *pCtg = NULL;
D
dapan1121 已提交
1513

dengyihao's avatar
dengyihao 已提交
1514
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
1515
  while (pIter) {
dengyihao's avatar
dengyihao 已提交
1516
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
1517 1518 1519

    if (pCtg) {
      ctgFreeHandle(pCtg);
1520 1521 1522 1523 1524 1525 1526 1527
    }

    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
  }

  taosHashClear(gCtgMgmt.pCluster);
}

D
dapan1121 已提交
1528
int32_t ctgOpUpdateVgroup(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1529
  int32_t          code = 0;
D
dapan1121 已提交
1530
  SCtgUpdateVgMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1531 1532 1533 1534
  SDBVgInfo       *dbInfo = msg->dbInfo;
  char            *dbFName = msg->dbFName;
  SCatalog        *pCtg = msg->pCtg;

D
dapan1121 已提交
1535
  if (NULL == dbInfo->vgHash) {
D
dapan1121 已提交
1536
    goto _return;
D
dapan1121 已提交
1537
  }
dengyihao's avatar
dengyihao 已提交
1538

D
dapan1121 已提交
1539
  if (dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgHash) <= 0) {
dengyihao's avatar
dengyihao 已提交
1540 1541
    ctgError("invalid db vgInfo, dbFName:%s, vgHash:%p, vgVersion:%d, vgHashSize:%d", dbFName, dbInfo->vgHash,
             dbInfo->vgVersion, taosHashGetSize(dbInfo->vgHash));
D
dapan1121 已提交
1542
    CTG_ERR_JRET(TSDB_CODE_APP_ERROR);
D
dapan1121 已提交
1543 1544
  }

dengyihao's avatar
dengyihao 已提交
1545
  bool         newAdded = false;
D
dapan1121 已提交
1546 1547 1548
  SDbVgVersion vgVersion = {.dbId = msg->dbId, .vgVersion = dbInfo->vgVersion, .numOfTable = dbInfo->numOfTable};

  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
1549
  CTG_ERR_JRET(ctgGetAddDBCache(msg->pCtg, dbFName, msg->dbId, &dbCache));
D
dapan1121 已提交
1550
  if (NULL == dbCache) {
dengyihao's avatar
dengyihao 已提交
1551
    ctgInfo("conflict db update, ignore this update, dbFName:%s, dbId:0x%" PRIx64, dbFName, msg->dbId);
D
dapan1121 已提交
1552
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1553 1554 1555
  }

  SCtgVgCache *vgCache = &dbCache->vgCache;
D
dapan1121 已提交
1556
  CTG_ERR_JRET(ctgWLockVgInfo(msg->pCtg, dbCache));
dengyihao's avatar
dengyihao 已提交
1557

D
dapan1121 已提交
1558 1559
  if (vgCache->vgInfo) {
    SDBVgInfo *vgInfo = vgCache->vgInfo;
dengyihao's avatar
dengyihao 已提交
1560

D
dapan1121 已提交
1561 1562 1563
    if (dbInfo->vgVersion < vgInfo->vgVersion) {
      ctgDebug("db vgVer is old, dbFName:%s, vgVer:%d, curVer:%d", dbFName, dbInfo->vgVersion, vgInfo->vgVersion);
      ctgWUnlockVgInfo(dbCache);
dengyihao's avatar
dengyihao 已提交
1564

D
dapan1121 已提交
1565
      goto _return;
D
dapan1121 已提交
1566 1567 1568
    }

    if (dbInfo->vgVersion == vgInfo->vgVersion && dbInfo->numOfTable == vgInfo->numOfTable) {
dengyihao's avatar
dengyihao 已提交
1569 1570
      ctgDebug("no new db vgVer or numOfTable, dbFName:%s, vgVer:%d, numOfTable:%d", dbFName, dbInfo->vgVersion,
               dbInfo->numOfTable);
D
dapan1121 已提交
1571
      ctgWUnlockVgInfo(dbCache);
dengyihao's avatar
dengyihao 已提交
1572

D
dapan1121 已提交
1573
      goto _return;
D
dapan1121 已提交
1574 1575 1576 1577 1578 1579 1580 1581
    }

    ctgFreeVgInfo(vgInfo);
  }

  vgCache->vgInfo = dbInfo;
  msg->dbInfo = NULL;

dengyihao's avatar
dengyihao 已提交
1582
  ctgDebug("db vgInfo updated, dbFName:%s, vgVer:%d, dbId:0x%" PRIx64, dbFName, vgVersion.vgVersion, vgVersion.dbId);
D
dapan1121 已提交
1583 1584 1585 1586 1587 1588

  ctgWUnlockVgInfo(dbCache);

  dbCache = NULL;

  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
dengyihao's avatar
dengyihao 已提交
1589 1590
  CTG_ERR_JRET(ctgMetaRentUpdate(&msg->pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion),
                                 ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1591 1592 1593 1594 1595

_return:

  ctgFreeVgInfo(msg->dbInfo);
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1596

D
dapan1121 已提交
1597 1598 1599
  CTG_RET(code);
}

D
dapan1121 已提交
1600
int32_t ctgOpDropDbCache(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1601
  int32_t        code = 0;
D
dapan1121 已提交
1602
  SCtgDropDBMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1603
  SCatalog      *pCtg = msg->pCtg;
D
dapan1121 已提交
1604 1605 1606 1607 1608 1609

  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
  if (NULL == dbCache) {
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
1610

D
dapan1121 已提交
1611
  if (dbCache->dbId != msg->dbId) {
dengyihao's avatar
dengyihao 已提交
1612 1613
    ctgInfo("dbId already updated, dbFName:%s, dbId:0x%" PRIx64 ", targetId:0x%" PRIx64, msg->dbFName, dbCache->dbId,
            msg->dbId);
D
dapan1121 已提交
1614 1615
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
1616

D
dapan1121 已提交
1617 1618 1619 1620 1621
  CTG_ERR_JRET(ctgRemoveDBFromCache(pCtg, dbCache, msg->dbFName));

_return:

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1622

D
dapan1121 已提交
1623 1624 1625
  CTG_RET(code);
}

D
dapan1121 已提交
1626
int32_t ctgOpDropDbVgroup(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1627
  int32_t              code = 0;
D
dapan1121 已提交
1628
  SCtgDropDbVgroupMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1629
  SCatalog            *pCtg = msg->pCtg;
D
dapan1121 已提交
1630 1631 1632 1633 1634 1635

  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
  if (NULL == dbCache) {
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
1636

dengyihao's avatar
dengyihao 已提交
1637
  CTG_ERR_JRET(ctgWLockVgInfo(pCtg, dbCache));
dengyihao's avatar
dengyihao 已提交
1638

D
dapan1121 已提交
1639 1640
  ctgFreeVgInfo(dbCache->vgCache.vgInfo);
  dbCache->vgCache.vgInfo = NULL;
D
dapan1121 已提交
1641 1642 1643

  ctgDebug("db vgInfo removed, dbFName:%s", msg->dbFName);

D
dapan1121 已提交
1644
  ctgWUnlockVgInfo(dbCache);
D
dapan1121 已提交
1645 1646 1647 1648

_return:

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1649

D
dapan1121 已提交
1650 1651 1652
  CTG_RET(code);
}

D
dapan1121 已提交
1653
int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1654
  int32_t              code = 0;
D
dapan1121 已提交
1655
  SCtgUpdateTbMetaMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1656 1657 1658
  SCatalog            *pCtg = msg->pCtg;
  STableMetaOutput    *pMeta = msg->pMeta;
  SCtgDBCache         *dbCache = NULL;
D
dapan1121 已提交
1659

D
dapan1121 已提交
1660 1661
  if ((!CTG_IS_META_CTABLE(pMeta->metaType)) && NULL == pMeta->tbMeta) {
    ctgError("no valid tbmeta got from meta rsp, dbFName:%s, tbName:%s", pMeta->dbFName, pMeta->tbName);
D
dapan1121 已提交
1662 1663 1664
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

D
dapan1121 已提交
1665 1666
  if (CTG_IS_META_BOTH(pMeta->metaType) && TSDB_SUPER_TABLE != pMeta->tbMeta->tableType) {
    ctgError("table type error, expected:%d, actual:%d", TSDB_SUPER_TABLE, pMeta->tbMeta->tableType);
D
dapan1121 已提交
1667
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
dengyihao's avatar
dengyihao 已提交
1668 1669
  }

D
dapan1121 已提交
1670
  CTG_ERR_JRET(ctgGetAddDBCache(pCtg, pMeta->dbFName, pMeta->dbId, &dbCache));
D
dapan1121 已提交
1671
  if (NULL == dbCache) {
D
dapan1121 已提交
1672
    ctgInfo("conflict db update, ignore this update, dbFName:%s, dbId:0x%" PRIx64, pMeta->dbFName, pMeta->dbId);
D
dapan1121 已提交
1673 1674 1675
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

D
dapan1121 已提交
1676 1677
  if (CTG_IS_META_TABLE(pMeta->metaType) || CTG_IS_META_BOTH(pMeta->metaType)) {
    int32_t metaSize = CTG_META_SIZE(pMeta->tbMeta);
dengyihao's avatar
dengyihao 已提交
1678 1679
    CTG_ERR_JRET(
        ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->tbName, pMeta->tbMeta, metaSize));
D
dapan1121 已提交
1680
    pMeta->tbMeta = NULL;
D
dapan1121 已提交
1681 1682
  }

D
dapan1121 已提交
1683
  if (CTG_IS_META_CTABLE(pMeta->metaType) || CTG_IS_META_BOTH(pMeta->metaType)) {
dengyihao's avatar
dengyihao 已提交
1684
    SCTableMeta *ctbMeta = taosMemoryMalloc(sizeof(SCTableMeta));
D
dapan1121 已提交
1685 1686 1687 1688
    if (NULL == ctbMeta) {
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
    }
    memcpy(ctbMeta, &pMeta->ctbMeta, sizeof(SCTableMeta));
dengyihao's avatar
dengyihao 已提交
1689 1690
    CTG_ERR_JRET(ctgWriteTbMetaToCache(pCtg, dbCache, pMeta->dbFName, pMeta->dbId, pMeta->ctbName,
                                       (STableMeta *)ctbMeta, sizeof(SCTableMeta)));
D
dapan1121 已提交
1691 1692 1693 1694
  }

_return:

D
dapan1121 已提交
1695 1696 1697
  if (pMeta) {
    taosMemoryFreeClear(pMeta->tbMeta);
    taosMemoryFreeClear(pMeta);
D
dapan1121 已提交
1698
  }
dengyihao's avatar
dengyihao 已提交
1699

D
dapan1121 已提交
1700
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1701

D
dapan1121 已提交
1702 1703 1704
  CTG_RET(code);
}

D
dapan1121 已提交
1705
int32_t ctgOpDropStbMeta(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1706
  int32_t             code = 0;
D
dapan1121 已提交
1707
  SCtgDropStbMetaMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1708
  SCatalog           *pCtg = msg->pCtg;
D
dapan1121 已提交
1709 1710 1711 1712

  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(pCtg, msg->dbFName, &dbCache);
  if (NULL == dbCache) {
dengyihao's avatar
dengyihao 已提交
1713
    goto _return;
D
dapan1121 已提交
1714 1715 1716
  }

  if (msg->dbId && (dbCache->dbId != msg->dbId)) {
dengyihao's avatar
dengyihao 已提交
1717
    ctgDebug("dbId already modified, dbFName:%s, current:0x%" PRIx64 ", dbId:0x%" PRIx64 ", stb:%s, suid:0x%" PRIx64,
D
dapan1121 已提交
1718
             msg->dbFName, dbCache->dbId, msg->dbId, msg->stbName, msg->suid);
dengyihao's avatar
dengyihao 已提交
1719
    goto _return;
D
dapan1121 已提交
1720
  }
dengyihao's avatar
dengyihao 已提交
1721

D
dapan1121 已提交
1722
  if (taosHashRemove(dbCache->stbCache, &msg->suid, sizeof(msg->suid))) {
dengyihao's avatar
dengyihao 已提交
1723 1724
    ctgDebug("stb not exist in stbCache, may be removed, dbFName:%s, stb:%s, suid:0x%" PRIx64, msg->dbFName,
             msg->stbName, msg->suid);
D
dapan1121 已提交
1725
  } else {
D
dapan1121 已提交
1726
    CTG_CACHE_STAT_DEC(numOfStb, 1);
D
dapan1121 已提交
1727 1728
  }

dengyihao's avatar
dengyihao 已提交
1729
  SCtgTbCache *pTbCache = taosHashGet(dbCache->tbCache, msg->stbName, strlen(msg->stbName));
D
dapan1121 已提交
1730 1731 1732 1733 1734 1735 1736 1737 1738
  if (NULL == pTbCache) {
    ctgDebug("stb %s already not in cache", msg->stbName);
    goto _return;
  }

  CTG_LOCK(CTG_WRITE, &pTbCache->metaLock);
  ctgFreeTbCacheImpl(pTbCache);
  CTG_UNLOCK(CTG_WRITE, &pTbCache->metaLock);

dengyihao's avatar
dengyihao 已提交
1739 1740
  if (taosHashRemove(dbCache->tbCache, msg->stbName, strlen(msg->stbName))) {
    ctgError("stb not exist in cache, dbFName:%s, stb:%s, suid:0x%" PRIx64, msg->dbFName, msg->stbName, msg->suid);
D
dapan1121 已提交
1741
  } else {
D
dapan1121 已提交
1742
    CTG_CACHE_STAT_DEC(numOfTbl, 1);
D
dapan1121 已提交
1743
  }
dengyihao's avatar
dengyihao 已提交
1744 1745

  ctgInfo("stb removed from cache, dbFName:%s, stbName:%s, suid:0x%" PRIx64, msg->dbFName, msg->stbName, msg->suid);
D
dapan1121 已提交
1746 1747

  CTG_ERR_JRET(ctgMetaRentRemove(&msg->pCtg->stbRent, msg->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
dengyihao's avatar
dengyihao 已提交
1748 1749 1750

  ctgDebug("stb removed from rent, dbFName:%s, stbName:%s, suid:0x%" PRIx64, msg->dbFName, msg->stbName, msg->suid);

D
dapan1121 已提交
1751 1752 1753
_return:

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1754

D
dapan1121 已提交
1755 1756 1757
  CTG_RET(code);
}

D
dapan1121 已提交
1758
int32_t ctgOpDropTbMeta(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1759
  int32_t             code = 0;
D
dapan1121 已提交
1760
  SCtgDropTblMetaMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1761
  SCatalog           *pCtg = msg->pCtg;
D
dapan1121 已提交
1762 1763 1764 1765

  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(pCtg, msg->dbFName, &dbCache);
  if (NULL == dbCache) {
D
dapan1121 已提交
1766
    goto _return;
D
dapan1121 已提交
1767 1768 1769
  }

  if (dbCache->dbId != msg->dbId) {
dengyihao's avatar
dengyihao 已提交
1770 1771
    ctgDebug("dbId 0x%" PRIx64 " not match with curId 0x%" PRIx64 ", dbFName:%s, tbName:%s", msg->dbId, dbCache->dbId,
             msg->dbFName, msg->tbName);
D
dapan1121 已提交
1772
    goto _return;
D
dapan1121 已提交
1773 1774
  }

dengyihao's avatar
dengyihao 已提交
1775
  SCtgTbCache *pTbCache = taosHashGet(dbCache->tbCache, msg->tbName, strlen(msg->tbName));
D
dapan1121 已提交
1776 1777 1778 1779 1780 1781 1782 1783
  if (NULL == pTbCache) {
    ctgDebug("tb %s already not in cache", msg->tbName);
    goto _return;
  }

  CTG_LOCK(CTG_WRITE, &pTbCache->metaLock);
  ctgFreeTbCacheImpl(pTbCache);
  CTG_UNLOCK(CTG_WRITE, &pTbCache->metaLock);
dengyihao's avatar
dengyihao 已提交
1784

D
dapan1121 已提交
1785 1786 1787
  if (taosHashRemove(dbCache->tbCache, msg->tbName, strlen(msg->tbName))) {
    ctgError("tb %s not exist in cache, dbFName:%s", msg->tbName, msg->dbFName);
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1788
  } else {
D
dapan1121 已提交
1789
    CTG_CACHE_STAT_DEC(numOfTbl, 1);
D
dapan1121 已提交
1790 1791
  }

D
dapan1121 已提交
1792
  ctgDebug("table %s removed from cache, dbFName:%s", msg->tbName, msg->dbFName);
D
dapan1121 已提交
1793 1794 1795 1796 1797 1798 1799 1800

_return:

  taosMemoryFreeClear(msg);

  CTG_RET(code);
}

D
dapan1121 已提交
1801
int32_t ctgOpUpdateUser(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1802
  int32_t            code = 0;
D
dapan1121 已提交
1803
  SCtgUpdateUserMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1804 1805
  SCatalog          *pCtg = msg->pCtg;

D
dapan1121 已提交
1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848
  SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user));
  if (NULL == pUser) {
    SCtgUserAuth userAuth = {0};

    userAuth.version = msg->userAuth.version;
    userAuth.superUser = msg->userAuth.superAuth;
    userAuth.createdDbs = msg->userAuth.createdDbs;
    userAuth.readDbs = msg->userAuth.readDbs;
    userAuth.writeDbs = msg->userAuth.writeDbs;

    if (taosHashPut(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user), &userAuth, sizeof(userAuth))) {
      ctgError("taosHashPut user %s to cache failed", msg->userAuth.user);
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
    }

    taosMemoryFreeClear(msg);

    return TSDB_CODE_SUCCESS;
  }

  pUser->version = msg->userAuth.version;

  CTG_LOCK(CTG_WRITE, &pUser->lock);

  taosHashCleanup(pUser->createdDbs);
  pUser->createdDbs = msg->userAuth.createdDbs;
  msg->userAuth.createdDbs = NULL;

  taosHashCleanup(pUser->readDbs);
  pUser->readDbs = msg->userAuth.readDbs;
  msg->userAuth.readDbs = NULL;

  taosHashCleanup(pUser->writeDbs);
  pUser->writeDbs = msg->userAuth.writeDbs;
  msg->userAuth.writeDbs = NULL;

  CTG_UNLOCK(CTG_WRITE, &pUser->lock);

_return:

  taosHashCleanup(msg->userAuth.createdDbs);
  taosHashCleanup(msg->userAuth.readDbs);
  taosHashCleanup(msg->userAuth.writeDbs);
dengyihao's avatar
dengyihao 已提交
1849

D
dapan1121 已提交
1850
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1851

D
dapan1121 已提交
1852 1853 1854
  CTG_RET(code);
}

D
dapan1121 已提交
1855
int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1856
  int32_t             code = 0;
D
dapan1121 已提交
1857
  SCtgUpdateEpsetMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1858 1859
  SCatalog           *pCtg = msg->pCtg;

D
dapan1121 已提交
1860
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
1861
  CTG_ERR_JRET(ctgGetDBCache(pCtg, msg->dbFName, &dbCache));
D
dapan1121 已提交
1862 1863 1864 1865 1866
  if (NULL == dbCache) {
    ctgDebug("db %s not exist, ignore epset update", msg->dbFName);
    goto _return;
  }

D
dapan1121 已提交
1867 1868
  CTG_ERR_JRET(ctgWLockVgInfo(pCtg, dbCache));

dengyihao's avatar
dengyihao 已提交
1869
  SDBVgInfo *vgInfo = dbCache->vgCache.vgInfo;
D
dapan1121 已提交
1870
  if (NULL == vgInfo) {
D
dapan1121 已提交
1871 1872 1873
    ctgDebug("vgroup in db %s not cached, ignore epset update", msg->dbFName);
    goto _return;
  }
dengyihao's avatar
dengyihao 已提交
1874 1875

  SVgroupInfo *pInfo = taosHashGet(vgInfo->vgHash, &msg->vgId, sizeof(msg->vgId));
D
dapan1121 已提交
1876 1877 1878 1879 1880
  if (NULL == pInfo) {
    ctgDebug("no vgroup %d in db %s, ignore epset update", msg->vgId, msg->dbFName);
    goto _return;
  }

dengyihao's avatar
dengyihao 已提交
1881 1882 1883 1884 1885
  SEp *pOrigEp = &pInfo->epSet.eps[pInfo->epSet.inUse];
  SEp *pNewEp = &msg->epSet.eps[msg->epSet.inUse];
  ctgDebug("vgroup %d epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d, dbFName:%s in ctg", pInfo->vgId,
           pInfo->epSet.inUse, pInfo->epSet.numOfEps, pOrigEp->fqdn, pOrigEp->port, msg->epSet.inUse,
           msg->epSet.numOfEps, pNewEp->fqdn, pNewEp->port, msg->dbFName);
D
dapan1121 已提交
1886

D
dapan1121 已提交
1887
  pInfo->epSet = msg->epSet;
D
dapan1121 已提交
1888 1889 1890 1891

_return:

  if (dbCache) {
D
dapan1121 已提交
1892
    ctgWUnlockVgInfo(dbCache);
D
dapan1121 已提交
1893 1894 1895
  }

  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1896

D
dapan1121 已提交
1897 1898 1899
  CTG_RET(code);
}

D
dapan1121 已提交
1900
int32_t ctgOpUpdateTbIndex(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1901
  int32_t               code = 0;
D
dapan1121 已提交
1902
  SCtgUpdateTbIndexMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1903 1904 1905 1906
  SCatalog             *pCtg = msg->pCtg;
  STableIndex          *pIndex = msg->pIndex;
  SCtgDBCache          *dbCache = NULL;

D
dapan1121 已提交
1907
  CTG_ERR_JRET(ctgGetAddDBCache(pCtg, pIndex->dbFName, 0, &dbCache));
D
dapan1121 已提交
1908 1909 1910 1911 1912 1913 1914 1915 1916

  CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, pIndex->dbFName, pIndex->tbName, &pIndex));

_return:

  if (pIndex) {
    taosArrayDestroyEx(pIndex->pIndex, tFreeSTableIndexInfo);
    taosMemoryFreeClear(pIndex);
  }
dengyihao's avatar
dengyihao 已提交
1917

D
dapan1121 已提交
1918
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1919

D
dapan1121 已提交
1920 1921 1922 1923
  CTG_RET(code);
}

int32_t ctgOpDropTbIndex(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1924
  int32_t             code = 0;
D
dapan1121 已提交
1925
  SCtgDropTbIndexMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1926 1927 1928
  SCatalog           *pCtg = msg->pCtg;
  SCtgDBCache        *dbCache = NULL;

D
dapan1121 已提交
1929
  CTG_ERR_JRET(ctgGetDBCache(pCtg, msg->dbFName, &dbCache));
D
dapan1121 已提交
1930
  if (NULL == dbCache) {
D
dapan1121 已提交
1931 1932 1933
    return TSDB_CODE_SUCCESS;
  }

dengyihao's avatar
dengyihao 已提交
1934
  STableIndex *pIndex = taosMemoryCalloc(1, sizeof(STableIndex));
D
dapan1121 已提交
1935 1936
  if (NULL == pIndex) {
    CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
1937
  }
D
dapan1121 已提交
1938 1939 1940
  strcpy(pIndex->tbName, msg->tbName);
  strcpy(pIndex->dbFName, msg->dbFName);
  pIndex->version = -1;
D
dapan1121 已提交
1941

D
dapan1121 已提交
1942
  CTG_ERR_JRET(ctgWriteTbIndexToCache(pCtg, dbCache, pIndex->dbFName, pIndex->tbName, &pIndex));
D
dapan1121 已提交
1943 1944 1945 1946 1947 1948 1949

_return:

  if (pIndex) {
    taosArrayDestroyEx(pIndex->pIndex, tFreeSTableIndexInfo);
    taosMemoryFreeClear(pIndex);
  }
dengyihao's avatar
dengyihao 已提交
1950

D
dapan1121 已提交
1951
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1952

D
dapan1121 已提交
1953 1954 1955
  CTG_RET(code);
}

D
dapan1121 已提交
1956
int32_t ctgOpClearCache(SCtgCacheOperation *operation) {
dengyihao's avatar
dengyihao 已提交
1957
  int32_t            code = 0;
D
dapan1121 已提交
1958
  SCtgClearCacheMsg *msg = operation->data;
dengyihao's avatar
dengyihao 已提交
1959
  SCatalog          *pCtg = msg->pCtg;
D
dapan1121 已提交
1960

D
dapan1121 已提交
1961 1962
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);

D
dapan1121 已提交
1963
  if (pCtg) {
D
dapan1121 已提交
1964 1965 1966 1967 1968
    if (msg->freeCtg) {
      ctgFreeHandle(pCtg);
    } else {
      ctgClearHandle(pCtg);
    }
dengyihao's avatar
dengyihao 已提交
1969

D
dapan1121 已提交
1970 1971
    goto _return;
  }
D
dapan1121 已提交
1972 1973 1974 1975 1976 1977

  if (msg->freeCtg) {
    ctgFreeAllInstance();
  } else {
    ctgClearAllInstance();
  }
D
dapan1121 已提交
1978 1979

_return:
D
dapan1121 已提交
1980 1981

  CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.lock);
dengyihao's avatar
dengyihao 已提交
1982

D
dapan1121 已提交
1983
  taosMemoryFreeClear(msg);
dengyihao's avatar
dengyihao 已提交
1984

D
dapan1121 已提交
1985 1986 1987
  CTG_RET(code);
}

D
dapan1121 已提交
1988
void ctgCleanupCacheQueue(void) {
dengyihao's avatar
dengyihao 已提交
1989 1990
  SCtgQNode          *node = NULL;
  SCtgQNode          *nodeNext = NULL;
D
dapan1121 已提交
1991
  SCtgCacheOperation *op = NULL;
dengyihao's avatar
dengyihao 已提交
1992
  bool                stopQueue = false;
D
dapan1121 已提交
1993 1994 1995 1996 1997

  while (true) {
    node = gCtgMgmt.queue.head->next;
    while (node) {
      if (node->op) {
D
dapan1121 已提交
1998 1999 2000 2001 2002 2003
        op = node->op;
        if (op->stopQueue) {
          SCatalog *pCtg = ((SCtgUpdateMsgHeader *)op->data)->pCtg;
          ctgDebug("process [%s] operation", gCtgCacheOperation[op->opId].name);
          (*gCtgCacheOperation[op->opId].func)(op);
          stopQueue = true;
dengyihao's avatar
dengyihao 已提交
2004
          CTG_RT_STAT_INC(numOfOpDequeue, 1);
D
dapan1121 已提交
2005 2006
        } else {
          taosMemoryFree(op->data);
dengyihao's avatar
dengyihao 已提交
2007
          CTG_RT_STAT_INC(numOfOpAbort, 1);
D
dapan1121 已提交
2008
        }
dengyihao's avatar
dengyihao 已提交
2009

D
dapan1121 已提交
2010 2011
        if (op->syncOp) {
          tsem_post(&op->rspSem);
D
dapan1121 已提交
2012
        } else {
D
dapan1121 已提交
2013
          taosMemoryFree(op);
D
dapan1121 已提交
2014
        }
D
dapan1121 已提交
2015
      }
D
dapan1121 已提交
2016 2017 2018

      nodeNext = node->next;
      taosMemoryFree(node);
dengyihao's avatar
dengyihao 已提交
2019

D
dapan1121 已提交
2020
      node = nodeNext;
D
dapan1121 已提交
2021 2022
    }

D
dapan1121 已提交
2023
    if (!stopQueue) {
D
dapan1121 已提交
2024 2025 2026 2027
      taosUsleep(1);
    } else {
      break;
    }
D
dapan1121 已提交
2028 2029 2030 2031 2032 2033
  }

  taosMemoryFreeClear(gCtgMgmt.queue.head);
  gCtgMgmt.queue.tail = NULL;
}

dengyihao's avatar
dengyihao 已提交
2034
void *ctgUpdateThreadFunc(void *param) {
D
dapan1121 已提交
2035
  setThreadName("catalog");
2036

D
dapan1121 已提交
2037 2038 2039 2040 2041 2042
  qInfo("catalog update thread started");

  while (true) {
    if (tsem_wait(&gCtgMgmt.queue.reqSem)) {
      qError("ctg tsem_wait failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
    }
dengyihao's avatar
dengyihao 已提交
2043 2044

    if (atomic_load_8((int8_t *)&gCtgMgmt.exit)) {
D
dapan1121 已提交
2045
      ctgCleanupCacheQueue();
D
dapan1121 已提交
2046 2047 2048
      break;
    }

D
dapan1121 已提交
2049 2050 2051
    SCtgCacheOperation *operation = NULL;
    ctgDequeue(&operation);
    SCatalog *pCtg = ((SCtgUpdateMsgHeader *)operation->data)->pCtg;
D
dapan1121 已提交
2052

D
dapan1121 已提交
2053
    ctgDebug("process [%s] operation", gCtgCacheOperation[operation->opId].name);
dengyihao's avatar
dengyihao 已提交
2054

D
dapan1121 已提交
2055
    (*gCtgCacheOperation[operation->opId].func)(operation);
D
dapan1121 已提交
2056

D
dapan1121 已提交
2057
    if (operation->syncOp) {
D
dapan1121 已提交
2058
      tsem_post(&operation->rspSem);
D
dapan1121 已提交
2059 2060
    } else {
      taosMemoryFreeClear(operation);
D
dapan1121 已提交
2061 2062
    }

dengyihao's avatar
dengyihao 已提交
2063
    CTG_RT_STAT_INC(numOfOpDequeue, 1);
D
dapan1121 已提交
2064

D
dapan1121 已提交
2065
    ctgdShowCacheInfo();
D
dapan1121 已提交
2066 2067 2068 2069
    ctgdShowClusterCache(pCtg);
  }

  qInfo("catalog update thread stopped");
dengyihao's avatar
dengyihao 已提交
2070

D
dapan1121 已提交
2071 2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082
  return NULL;
}

int32_t ctgStartUpdateThread() {
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);

  if (taosThreadCreate(&gCtgMgmt.updateThread, &thAttr, ctgUpdateThreadFunc, NULL) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    CTG_ERR_RET(terrno);
  }
dengyihao's avatar
dengyihao 已提交
2083

D
dapan1121 已提交
2084 2085 2086 2087
  taosThreadAttrDestroy(&thAttr);
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
2088
int32_t ctgGetTbMetaFromCache(SCatalog *pCtg, SRequestConnInfo *pConn, SCtgTbMetaCtx *ctx, STableMeta **pTableMeta) {
D
dapan1121 已提交
2089
  if (IS_SYS_DBNAME(ctx->pName->dbname)) {
D
dapan1121 已提交
2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110
    CTG_FLAG_SET_SYS_DB(ctx->flag);
  }

  CTG_ERR_RET(ctgReadTbMetaFromCache(pCtg, ctx, pTableMeta));

  if (*pTableMeta) {
    if (CTG_FLAG_MATCH_STB(ctx->flag, (*pTableMeta)->tableType) &&
        ((!CTG_FLAG_IS_FORCE_UPDATE(ctx->flag)) || (CTG_FLAG_IS_SYS_DB(ctx->flag)))) {
      return TSDB_CODE_SUCCESS;
    }

    taosMemoryFreeClear(*pTableMeta);
  }

  if (CTG_FLAG_IS_UNKNOWN_STB(ctx->flag)) {
    CTG_FLAG_SET_STB(ctx->flag, ctx->tbInfo.tbType);
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2111
#if 0
2112
int32_t ctgGetTbMetaBFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx* ctx, SArray** pResList) {
D
dapan1121 已提交
2113 2114 2115 2116 2117 2118 2119 2120 2121 2122 2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149
  int32_t tbNum = taosArrayGetSize(ctx->pNames);
  SName* fName = taosArrayGet(ctx->pNames, 0);
  int32_t fIdx = 0;
  
  for (int32_t i = 0; i < tbNum; ++i) {
    SName* pName = taosArrayGet(ctx->pNames, i);
    SCtgTbMetaCtx nctx = {0};
    nctx.flag = CTG_FLAG_UNKNOWN_STB;
    nctx.pName = pName;
    
    if (IS_SYS_DBNAME(pName->dbname)) {
      CTG_FLAG_SET_SYS_DB(nctx.flag);
    }

    STableMeta *pTableMeta = NULL;
    CTG_ERR_RET(ctgReadTbMetaFromCache(pCtg, &nctx, &pTableMeta));
    SMetaRes res = {0};
    
    if (pTableMeta) {
      if (CTG_FLAG_MATCH_STB(nctx.flag, pTableMeta->tableType) &&
          ((!CTG_FLAG_IS_FORCE_UPDATE(nctx.flag)) || (CTG_FLAG_IS_SYS_DB(nctx.flag)))) {
        res.pRes = pTableMeta;
      } else {
        taosMemoryFreeClear(pTableMeta);
      }
    }

    if (NULL == res.pRes) {
      if (NULL == ctx->pFetchs) {
        ctx->pFetchs = taosArrayInit(tbNum, sizeof(SCtgFetch));
      }
      
      if (CTG_FLAG_IS_UNKNOWN_STB(nctx.flag)) {
        CTG_FLAG_SET_STB(nctx.flag, nctx.tbInfo.tbType);
      }

      SCtgFetch fetch = {0};
2150
      fetch.tbIdx = i;
D
dapan1121 已提交
2151 2152 2153 2154 2155 2156
      fetch.fetchIdx = fIdx++;
      fetch.flag = nctx.flag;

      taosArrayPush(ctx->pFetchs, &fetch);
    }
    
2157
    taosArrayPush(ctx->pResList, &res);
D
dapan1121 已提交
2158 2159 2160
  }

  if (NULL == ctx->pFetchs) {
2161
    TSWAP(*pResList, ctx->pResList);
D
dapan1121 已提交
2162 2163 2164 2165
  }

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
2166 2167
#endif

dengyihao's avatar
dengyihao 已提交
2168 2169 2170 2171 2172 2173 2174 2175 2176
int32_t ctgGetTbMetasFromCache(SCatalog *pCtg, SRequestConnInfo *pConn, SCtgTbMetasCtx *ctx, int32_t dbIdx,
                               int32_t *fetchIdx, int32_t baseResIdx, SArray *pList) {
  int32_t     tbNum = taosArrayGetSize(pList);
  SName      *pName = taosArrayGet(pList, 0);
  char        dbFName[TSDB_DB_FNAME_LEN] = {0};
  int32_t     flag = CTG_FLAG_UNKNOWN_STB;
  uint64_t    lastSuid = 0;
  STableMeta *lastTableMeta = NULL;

D
dapan1121 已提交
2177 2178 2179 2180 2181 2182 2183 2184
  if (IS_SYS_DBNAME(pName->dbname)) {
    CTG_FLAG_SET_SYS_DB(flag);
    strcpy(dbFName, pName->dbname);
  } else {
    tNameGetFullDbName(pName, dbFName);
  }

  SCtgDBCache *dbCache = NULL;
dengyihao's avatar
dengyihao 已提交
2185
  SCtgTbCache *pCache = NULL;
D
dapan1121 已提交
2186
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
dengyihao's avatar
dengyihao 已提交
2187

D
dapan1121 已提交
2188 2189 2190
  if (NULL == dbCache) {
    ctgDebug("db %s not in cache", dbFName);
    for (int32_t i = 0; i < tbNum; ++i) {
2191 2192
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
D
dapan1121 已提交
2193 2194 2195 2196 2197 2198
    }

    return TSDB_CODE_SUCCESS;
  }

  for (int32_t i = 0; i < tbNum; ++i) {
dengyihao's avatar
dengyihao 已提交
2199
    SName *pName = taosArrayGet(pList, i);
D
dapan1121 已提交
2200 2201 2202 2203

    pCache = taosHashAcquire(dbCache->tbCache, pName->tname, strlen(pName->tname));
    if (NULL == pCache) {
      ctgDebug("tb %s not in cache, dbFName:%s", pName->tname, dbFName);
2204 2205
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
dengyihao's avatar
dengyihao 已提交
2206

D
dapan1121 已提交
2207 2208 2209 2210 2211 2212
      continue;
    }

    CTG_LOCK(CTG_READ, &pCache->metaLock);
    if (NULL == pCache->pMeta) {
      ctgDebug("tb %s meta not in cache, dbFName:%s", pName->tname, dbFName);
2213 2214
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
dengyihao's avatar
dengyihao 已提交
2215

D
dapan1121 已提交
2216 2217 2218
      continue;
    }

dengyihao's avatar
dengyihao 已提交
2219
    STableMeta *tbMeta = pCache->pMeta;
D
dapan1121 已提交
2220 2221 2222 2223 2224 2225 2226 2227

    SCtgTbMetaCtx nctx = {0};
    nctx.flag = flag;
    nctx.tbInfo.inCache = true;
    nctx.tbInfo.dbId = dbCache->dbId;
    nctx.tbInfo.suid = tbMeta->suid;
    nctx.tbInfo.tbType = tbMeta->tableType;

dengyihao's avatar
dengyihao 已提交
2228 2229
    SMetaRes    res = {0};
    STableMeta *pTableMeta = NULL;
D
dapan1121 已提交
2230 2231 2232 2233
    if (tbMeta->tableType != TSDB_CHILD_TABLE) {
      int32_t metaSize = CTG_META_SIZE(tbMeta);
      pTableMeta = taosMemoryCalloc(1, metaSize);
      if (NULL == pTableMeta) {
2234
        ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);
D
dapan1121 已提交
2235 2236
        CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
      }
dengyihao's avatar
dengyihao 已提交
2237

D
dapan1121 已提交
2238
      memcpy(pTableMeta, tbMeta, metaSize);
dengyihao's avatar
dengyihao 已提交
2239

2240
      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
2241 2242
      taosHashRelease(dbCache->tbCache, pCache);

D
dapan1121 已提交
2243
      ctgDebug("Got tb %s meta from cache, type:%d, dbFName:%s", pName->tname, tbMeta->tableType, dbFName);
dengyihao's avatar
dengyihao 已提交
2244

D
dapan1121 已提交
2245
      res.pRes = pTableMeta;
2246
      taosArrayPush(ctx->pResList, &res);
D
dapan1121 已提交
2247 2248 2249

      continue;
    }
dengyihao's avatar
dengyihao 已提交
2250

D
dapan1121 已提交
2251 2252 2253 2254
    // PROCESS FOR CHILD TABLE

    if (lastSuid && tbMeta->suid == lastSuid && lastTableMeta) {
      cloneTableMeta(lastTableMeta, &pTableMeta);
2255
      memcpy(pTableMeta, tbMeta, sizeof(SCTableMeta));
D
dapan1121 已提交
2256

2257
      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
2258 2259
      taosHashRelease(dbCache->tbCache, pCache);

D
dapan1121 已提交
2260
      ctgDebug("Got tb %s meta from cache, type:%d, dbFName:%s", pName->tname, tbMeta->tableType, dbFName);
dengyihao's avatar
dengyihao 已提交
2261

D
dapan1121 已提交
2262
      res.pRes = pTableMeta;
2263
      taosArrayPush(ctx->pResList, &res);
dengyihao's avatar
dengyihao 已提交
2264

D
dapan1121 已提交
2265 2266
      continue;
    }
dengyihao's avatar
dengyihao 已提交
2267

D
dapan1121 已提交
2268 2269 2270
    int32_t metaSize = sizeof(SCTableMeta);
    pTableMeta = taosMemoryCalloc(1, metaSize);
    if (NULL == pTableMeta) {
dengyihao's avatar
dengyihao 已提交
2271
      ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);
D
dapan1121 已提交
2272 2273
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
    }
dengyihao's avatar
dengyihao 已提交
2274

D
dapan1121 已提交
2275
    memcpy(pTableMeta, tbMeta, metaSize);
dengyihao's avatar
dengyihao 已提交
2276

2277
    CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
2278 2279 2280 2281 2282 2283
    taosHashRelease(dbCache->tbCache, pCache);

    ctgDebug("Got ctb %s meta from cache, will continue to get its stb meta, type:%d, dbFName:%s", pName->tname,
             nctx.tbInfo.tbType, dbFName);

    char *stName = taosHashAcquire(dbCache->stbCache, &pTableMeta->suid, sizeof(pTableMeta->suid));
D
dapan1121 已提交
2284 2285
    if (NULL == stName) {
      ctgDebug("stb 0x%" PRIx64 " not in cache, dbFName:%s", pTableMeta->suid, dbFName);
2286 2287
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
D
dapan1121 已提交
2288 2289 2290 2291 2292 2293 2294 2295 2296

      taosMemoryFreeClear(pTableMeta);
      continue;
    }

    pCache = taosHashAcquire(dbCache->tbCache, stName, strlen(stName));
    if (NULL == pCache) {
      ctgDebug("stb 0x%" PRIx64 " name %s not in cache, dbFName:%s", pTableMeta->suid, stName, dbFName);
      taosHashRelease(dbCache->stbCache, stName);
dengyihao's avatar
dengyihao 已提交
2297

2298 2299
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
D
dapan1121 已提交
2300

dengyihao's avatar
dengyihao 已提交
2301
      taosMemoryFreeClear(pTableMeta);
D
dapan1121 已提交
2302 2303 2304 2305 2306 2307 2308 2309
      continue;
    }

    taosHashRelease(dbCache->stbCache, stName);

    CTG_LOCK(CTG_READ, &pCache->metaLock);
    if (NULL == pCache->pMeta) {
      ctgDebug("stb 0x%" PRIx64 " meta not in cache, dbFName:%s", pTableMeta->suid, dbFName);
2310
      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
2311 2312
      taosHashRelease(dbCache->tbCache, pCache);

2313 2314
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
D
dapan1121 已提交
2315 2316 2317 2318 2319

      taosMemoryFreeClear(pTableMeta);

      continue;
    }
dengyihao's avatar
dengyihao 已提交
2320 2321 2322

    STableMeta *stbMeta = pCache->pMeta;
    if (stbMeta->suid != nctx.tbInfo.suid) {
2323
      CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
2324 2325 2326 2327 2328
      taosHashRelease(dbCache->tbCache, pCache);

      ctgError("stb suid 0x%" PRIx64 " in stbCache mis-match, expected suid 0x%" PRIx64, stbMeta->suid,
               nctx.tbInfo.suid);

2329 2330
      ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag);
      taosArraySetSize(ctx->pResList, taosArrayGetSize(ctx->pResList) + 1);
D
dapan1121 已提交
2331 2332 2333 2334 2335

      taosMemoryFreeClear(pTableMeta);

      continue;
    }
dengyihao's avatar
dengyihao 已提交
2336

D
dapan1121 已提交
2337 2338
    metaSize = CTG_META_SIZE(stbMeta);
    pTableMeta = taosMemoryRealloc(pTableMeta, metaSize);
dengyihao's avatar
dengyihao 已提交
2339
    if (NULL == pTableMeta) {
2340
      ctgReleaseTbMetaToCache(pCtg, dbCache, pCache);
D
dapan1121 已提交
2341 2342
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
    }
dengyihao's avatar
dengyihao 已提交
2343

D
dapan1121 已提交
2344
    memcpy(&pTableMeta->sversion, &stbMeta->sversion, metaSize - sizeof(SCTableMeta));
dengyihao's avatar
dengyihao 已提交
2345

2346
    CTG_UNLOCK(CTG_READ, &pCache->metaLock);
dengyihao's avatar
dengyihao 已提交
2347 2348
    taosHashRelease(dbCache->tbCache, pCache);

D
dapan1121 已提交
2349
    res.pRes = pTableMeta;
2350
    taosArrayPush(ctx->pResList, &res);
D
dapan1121 已提交
2351 2352 2353 2354 2355

    lastSuid = pTableMeta->suid;
    lastTableMeta = pTableMeta;
  }

2356
  ctgReleaseDBCache(pCtg, dbCache);
dengyihao's avatar
dengyihao 已提交
2357

D
dapan1121 已提交
2358 2359 2360
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
2361
int32_t ctgRemoveTbMetaFromCache(SCatalog *pCtg, SName *pTableName, bool syncReq) {
D
dapan1121 已提交
2362
  int32_t       code = 0;
dengyihao's avatar
dengyihao 已提交
2363
  STableMeta   *tblMeta = NULL;
D
dapan1121 已提交
2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391
  SCtgTbMetaCtx tbCtx = {0};
  tbCtx.flag = CTG_FLAG_UNKNOWN_STB;
  tbCtx.pName = pTableName;

  CTG_ERR_JRET(ctgReadTbMetaFromCache(pCtg, &tbCtx, &tblMeta));

  if (NULL == tblMeta) {
    ctgDebug("table already not in cache, db:%s, tblName:%s", pTableName->dbname, pTableName->tname);
    return TSDB_CODE_SUCCESS;
  }

  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);

  if (TSDB_SUPER_TABLE == tblMeta->tableType) {
    CTG_ERR_JRET(ctgDropStbMetaEnqueue(pCtg, dbFName, tbCtx.tbInfo.dbId, pTableName->tname, tblMeta->suid, syncReq));
  } else {
    CTG_ERR_JRET(ctgDropTbMetaEnqueue(pCtg, dbFName, tbCtx.tbInfo.dbId, pTableName->tname, syncReq));
  }

_return:

  taosMemoryFreeClear(tblMeta);

  CTG_RET(code);
}

int32_t ctgGetTbHashVgroupFromCache(SCatalog *pCtg, const SName *pTableName, SVgroupInfo **pVgroup) {
D
dapan1121 已提交
2392
  if (IS_SYS_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
2393 2394 2395 2396
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

dengyihao's avatar
dengyihao 已提交
2397
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423
  int32_t      code = 0;
  char         dbFName[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, dbFName);

  CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));

  if (NULL == dbCache) {
    *pVgroup = NULL;
    return TSDB_CODE_SUCCESS;
  }

  *pVgroup = taosMemoryCalloc(1, sizeof(SVgroupInfo));
  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pTableName, *pVgroup));

_return:

  if (dbCache) {
    ctgReleaseVgInfoToCache(pCtg, dbCache);
  }

  if (code) {
    taosMemoryFreeClear(*pVgroup);
  }

  CTG_RET(code);
}