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

#include "trpc.h"
#include "query.h"
#include "tname.h"
#include "catalogInt.h"
#include "systable.h"

D
dapan1121 已提交
22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44

void ctgFreeBatch(SCtgBatch *pBatch) {
  if (NULL == pBatch) {
    return;
  }
  
  taosArrayDestroy(pBatch->pMsgs);
  taosArrayDestroy(pBatch->pTaskIds);
}

void ctgFreeBatchs(SHashObj *pBatchs) {
  void* p = taosHashIterate(pBatchs, NULL);
  while (NULL != p) {
    SCtgBatch* pBatch = (SCtgBatch*)p;

    ctgFreeBatch(pBatch);

    p = taosHashIterate(pBatchs, p);
  }

  taosHashCleanup(pBatchs);
}

D
dapan1121 已提交
45 46 47 48
char *ctgTaskTypeStr(CTG_TASK_TYPE type) {
  switch (type) {
    case CTG_TASK_GET_QNODE:
      return "[get qnode list]";
D
dapan1121 已提交
49 50
    case CTG_TASK_GET_DNODE:
      return "[get dnode list]";
D
dapan1121 已提交
51 52 53 54 55 56 57 58 59 60
    case CTG_TASK_GET_DB_VGROUP:
      return "[get db vgroup]";
    case CTG_TASK_GET_DB_CFG:
      return "[get db cfg]";
    case CTG_TASK_GET_DB_INFO:
      return "[get db info]";
    case CTG_TASK_GET_TB_META:
      return "[get table meta]";
    case CTG_TASK_GET_TB_HASH:
      return "[get table hash]";
D
dapan1121 已提交
61 62 63 64
    case CTG_TASK_GET_TB_INDEX:
      return "[get table index]";
    case CTG_TASK_GET_TB_CFG:
      return "[get table cfg]";
D
dapan1121 已提交
65 66 67 68 69 70
    case CTG_TASK_GET_INDEX:
      return "[get index]";
    case CTG_TASK_GET_UDF:
      return "[get udf]";
    case CTG_TASK_GET_USER:
      return "[get user]";
D
dapan1121 已提交
71 72
    case CTG_TASK_GET_SVR_VER:
      return "[get svr ver]";
D
dapan1121 已提交
73 74 75 76 77
    default:
      return "unknown";
  }
}

78 79 80 81
void ctgFreeQNode(SCtgQNode *node) {
  //TODO
}

D
dapan1121 已提交
82 83 84 85 86 87 88 89 90 91
void ctgFreeSTableIndex(void *info) {
  if (NULL == info) {
    return;
  }

  STableIndex *pInfo = (STableIndex *)info;

  taosArrayDestroyEx(pInfo->pIndex, tFreeSTableIndexInfo);
}

D
dapan1121 已提交
92 93 94
void ctgFreeSMetaData(SMetaData* pData) {
  taosArrayDestroy(pData->pTableMeta);
  pData->pTableMeta = NULL;
D
dapan1121 已提交
95 96

/*  
D
dapan1121 已提交
97 98 99 100
  for (int32_t i = 0; i < taosArrayGetSize(pData->pDbVgroup); ++i) {
    SArray** pArray = taosArrayGet(pData->pDbVgroup, i);
    taosArrayDestroy(*pArray);
  }
D
dapan1121 已提交
101
*/
D
dapan1121 已提交
102 103 104 105 106
  taosArrayDestroy(pData->pDbVgroup);
  pData->pDbVgroup = NULL;
  
  taosArrayDestroy(pData->pTableHash);
  pData->pTableHash = NULL;
D
dapan1121 已提交
107 108 109

  taosArrayDestroy(pData->pTableIndex);
  pData->pTableIndex = NULL;
D
dapan1121 已提交
110 111 112
  
  taosArrayDestroy(pData->pUdfList);
  pData->pUdfList = NULL;
D
dapan1121 已提交
113

D
dapan1121 已提交
114
/*
D
dapan1121 已提交
115 116 117 118
  for (int32_t i = 0; i < taosArrayGetSize(pData->pDbCfg); ++i) {
    SDbCfgInfo* pInfo = taosArrayGet(pData->pDbCfg, i);
    taosArrayDestroy(pInfo->pRetensions);
  }
D
dapan1121 已提交
119
*/  
D
dapan1121 已提交
120 121
  taosArrayDestroy(pData->pDbCfg);
  pData->pDbCfg = NULL;
D
dapan1121 已提交
122 123 124

  taosArrayDestroy(pData->pDbInfo);
  pData->pDbInfo = NULL;
D
dapan1121 已提交
125 126 127 128 129 130 131 132 133
  
  taosArrayDestroy(pData->pIndex);
  pData->pIndex = NULL;
  
  taosArrayDestroy(pData->pUser);
  pData->pUser = NULL;
  
  taosArrayDestroy(pData->pQnodeList);
  pData->pQnodeList = NULL;
D
dapan1121 已提交
134

D
dapan1121 已提交
135 136 137
  taosArrayDestroy(pData->pDnodeList);
  pData->pDnodeList = NULL;

D
dapan1121 已提交
138 139
  taosArrayDestroy(pData->pTableCfg);
  pData->pTableCfg = NULL;
D
dapan1121 已提交
140 141

  taosMemoryFreeClear(pData->pSvrVer);
D
dapan1121 已提交
142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165
}

void ctgFreeSCtgUserAuth(SCtgUserAuth *userCache) {
  taosHashCleanup(userCache->createdDbs);
  taosHashCleanup(userCache->readDbs);
  taosHashCleanup(userCache->writeDbs);
}

void ctgFreeMetaRent(SCtgRentMgmt *mgmt) {
  if (NULL == mgmt->slots) {
    return;
  }

  for (int32_t i = 0; i < mgmt->slotNum; ++i) {
    SCtgRentSlot *slot = &mgmt->slots[i];
    if (slot->meta) {
      taosArrayDestroy(slot->meta);
      slot->meta = NULL;
    }
  }

  taosMemoryFreeClear(mgmt->slots);
}

D
dapan1121 已提交
166 167 168 169 170
void ctgFreeStbMetaCache(SCtgDBCache *dbCache) {
  if (NULL == dbCache->stbCache) {
    return;
  }

D
dapan1121 已提交
171
  int32_t stbNum = taosHashGetSize(dbCache->stbCache);  
D
dapan1121 已提交
172 173
  taosHashCleanup(dbCache->stbCache);
  dbCache->stbCache = NULL;
D
dapan1121 已提交
174
  CTG_CACHE_STAT_DEC(numOfStb, stbNum);
D
dapan1121 已提交
175 176 177
}

void ctgFreeTbCacheImpl(SCtgTbCache *pCache) {
D
dapan1121 已提交
178
  qDebug("tbMeta freed, p:%p", pCache->pMeta);
D
dapan1121 已提交
179 180 181 182 183 184
  taosMemoryFreeClear(pCache->pMeta);
  if (pCache->pIndex) {
    taosArrayDestroyEx(pCache->pIndex->pIndex, tFreeSTableIndexInfo);
    taosMemoryFreeClear(pCache->pIndex);
  }
}
D
dapan1121 已提交
185

D
dapan1121 已提交
186 187 188
void ctgFreeTbCache(SCtgDBCache *dbCache) {
  if (NULL == dbCache->tbCache) {
    return;
D
dapan1121 已提交
189 190
  }

D
dapan1121 已提交
191 192 193 194 195
  int32_t tblNum = taosHashGetSize(dbCache->tbCache);
  SCtgTbCache *pCache = taosHashIterate(dbCache->tbCache, NULL);
  while (NULL != pCache) {
    ctgFreeTbCacheImpl(pCache);
    pCache = taosHashIterate(dbCache->tbCache, pCache);
D
dapan1121 已提交
196
  }
D
dapan1121 已提交
197 198
  taosHashCleanup(dbCache->tbCache);
  dbCache->tbCache = NULL;
D
dapan1121 已提交
199
  CTG_CACHE_STAT_DEC(numOfTbl, tblNum);
D
dapan1121 已提交
200 201 202 203 204 205 206 207 208 209 210 211 212 213 214
}

void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
  if (NULL == vgInfo) {
    return;
  }

  if (vgInfo->vgHash) {
    taosHashCleanup(vgInfo->vgHash);
    vgInfo->vgHash = NULL;
  }
  
  taosMemoryFreeClear(vgInfo);
}

D
dapan1121 已提交
215 216 217 218
void ctgFreeVgInfoCache(SCtgDBCache *dbCache) {
  ctgFreeVgInfo(dbCache->vgCache.vgInfo);
}

D
dapan1121 已提交
219 220 221 222 223
void ctgFreeDbCache(SCtgDBCache *dbCache) {
  if (NULL == dbCache) {
    return;
  }

D
dapan1121 已提交
224 225 226
  ctgFreeVgInfoCache(dbCache);
  ctgFreeStbMetaCache(dbCache);
  ctgFreeTbCache(dbCache);
D
dapan1121 已提交
227 228
}

D
dapan1121 已提交
229 230 231 232
void ctgFreeInstDbCache(SHashObj* pDbCache) {
  if (NULL == pDbCache) {
    return;
  }
D
dapan1121 已提交
233
  
D
dapan1121 已提交
234 235 236 237 238 239 240 241 242 243 244 245 246 247 248
  int32_t dbNum = taosHashGetSize(pDbCache);
  
  void *pIter = taosHashIterate(pDbCache, NULL);
  while (pIter) {
    SCtgDBCache *dbCache = pIter;
    atomic_store_8(&dbCache->deleted, 1);
    ctgFreeDbCache(dbCache);
          
    pIter = taosHashIterate(pDbCache, pIter);
  }  

  taosHashCleanup(pDbCache);
  
  CTG_CACHE_STAT_DEC(numOfDb, dbNum);
}
D
dapan1121 已提交
249

D
dapan1121 已提交
250 251 252
void ctgFreeInstUserCache(SHashObj* pUserCache) {
  if (NULL == pUserCache) {
    return;
D
dapan1121 已提交
253
  }
D
dapan1121 已提交
254 255 256 257 258 259 260 261 262 263 264 265 266 267 268
  
  int32_t userNum = taosHashGetSize(pUserCache);
  
  void *pIter = taosHashIterate(pUserCache, NULL);
  while (pIter) {
    SCtgUserAuth *userCache = pIter;
    ctgFreeSCtgUserAuth(userCache);
  
    pIter = taosHashIterate(pUserCache, pIter);
  }  
  
  taosHashCleanup(pUserCache);
  
  CTG_CACHE_STAT_DEC(numOfUser, userNum);
}
D
dapan1121 已提交
269

D
dapan1121 已提交
270 271 272
void ctgFreeHandleImpl(SCatalog* pCtg) {
  ctgFreeMetaRent(&pCtg->dbRent);
  ctgFreeMetaRent(&pCtg->stbRent);
D
dapan1121 已提交
273

D
dapan1121 已提交
274 275
  ctgFreeInstDbCache(pCtg->dbCache);
  ctgFreeInstUserCache(pCtg->userCache);
D
dapan1121 已提交
276 277 278 279 280

  taosMemoryFree(pCtg);
}


D
dapan1121 已提交
281 282 283 284 285
void ctgFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
    return;
  }

D
dapan1121 已提交
286
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
287

D
dapan1121 已提交
288 289
  ctgFreeMetaRent(&pCtg->dbRent);
  ctgFreeMetaRent(&pCtg->stbRent);
D
dapan1121 已提交
290

D
dapan1121 已提交
291 292 293 294
  ctgFreeInstDbCache(pCtg->dbCache);
  ctgFreeInstUserCache(pCtg->userCache);

  CTG_CACHE_STAT_DEC(numOfCluster, 1);
D
dapan1121 已提交
295

D
dapan1121 已提交
296
  taosMemoryFree(pCtg);
D
dapan1121 已提交
297

C
Cary Xu 已提交
298
  ctgInfo("handle freed, clusterId:0x%" PRIx64, clusterId);
D
dapan1121 已提交
299 300
}

D
dapan1121 已提交
301 302 303 304
void ctgClearHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
    return;
  }
D
dapan1121 已提交
305

D
dapan1121 已提交
306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328
  uint64_t clusterId = pCtg->clusterId;

  ctgFreeMetaRent(&pCtg->dbRent);
  ctgFreeMetaRent(&pCtg->stbRent);

  ctgFreeInstDbCache(pCtg->dbCache);
  ctgFreeInstUserCache(pCtg->userCache);

  ctgMetaRentInit(&pCtg->dbRent, gCtgMgmt.cfg.dbRentSec, CTG_RENT_DB);
  ctgMetaRentInit(&pCtg->stbRent, gCtgMgmt.cfg.stbRentSec, CTG_RENT_STABLE);
  
  pCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
  if (NULL == pCtg->dbCache) {
    qError("taosHashInit %d dbCache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
  }
  
  pCtg->userCache = taosHashInit(gCtgMgmt.cfg.maxUserCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
  if (NULL == pCtg->userCache) {
    ctgError("taosHashInit %d user cache failed", gCtgMgmt.cfg.maxUserCacheNum);
  }

  CTG_CACHE_STAT_INC(numOfClear, 1);

C
Cary Xu 已提交
329
  ctgInfo("handle cleared, clusterId:0x%" PRIx64, clusterId);
D
dapan1121 已提交
330
}
D
dapan1121 已提交
331

D
dapan1121 已提交
332
void ctgFreeSUseDbOutput(SUseDbOutput* pOutput) {
D
dapan1121 已提交
333
  if (NULL == pOutput) {
D
dapan1121 已提交
334 335 336
    return;
  }

D
dapan1121 已提交
337 338 339 340 341
  if (pOutput->dbVgroup) {
    taosHashCleanup(pOutput->dbVgroup->vgHash);
    taosMemoryFreeClear(pOutput->dbVgroup);
  }
  
D
dapan1121 已提交
342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381
  taosMemoryFree(pOutput);
}

void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) {
  taosMemoryFreeClear(pCtx->target);
  if (NULL == pCtx->out) {
    return;
  }
  
  switch (pCtx->reqType) {
    case TDMT_MND_GET_DB_CFG: {
      SDbCfgInfo* pOut = (SDbCfgInfo*)pCtx->out;
      taosArrayDestroy(pOut->pRetensions);
      taosMemoryFreeClear(pCtx->out);
      break;
    }
    case TDMT_MND_USE_DB:{
      SUseDbOutput* pOut = (SUseDbOutput*)pCtx->out;
      ctgFreeSUseDbOutput(pOut);
      pCtx->out = NULL;
      break;
    }
    case TDMT_MND_GET_INDEX: {
      SIndexInfo* pOut = (SIndexInfo*)pCtx->out;
      taosMemoryFreeClear(pCtx->out);
      break;
    }
    case TDMT_MND_QNODE_LIST: {
      SArray* pOut = (SArray*)pCtx->out;
      taosArrayDestroy(pOut);
      pCtx->out = NULL;
      break;
    }
    case TDMT_VND_TABLE_META:
    case TDMT_MND_TABLE_META: {
      STableMetaOutput* pOut = (STableMetaOutput*)pCtx->out;
      taosMemoryFree(pOut->tbMeta);
      taosMemoryFreeClear(pCtx->out);
      break;
    }
D
dapan1121 已提交
382
    case TDMT_MND_GET_TABLE_INDEX: {
D
dapan1121 已提交
383
      STableIndex* pOut = (STableIndex*)pCtx->out;
D
dapan1121 已提交
384
      if (pOut) {
D
dapan1121 已提交
385
        taosArrayDestroyEx(pOut->pIndex, tFreeSTableIndexInfo);
D
dapan1121 已提交
386 387 388 389
        taosMemoryFreeClear(pCtx->out);
      }
      break;
    }
D
dapan1121 已提交
390 391 392 393 394 395 396
    case TDMT_VND_TABLE_CFG:
    case TDMT_MND_TABLE_CFG: {
      STableCfgRsp* pOut = (STableCfgRsp*)pCtx->out;
      tFreeSTableCfgRsp(pOut);
      taosMemoryFreeClear(pCtx->out);
      break;
    }
D
dapan1121 已提交
397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444
    case TDMT_MND_RETRIEVE_FUNC: {
      SFuncInfo* pOut = (SFuncInfo*)pCtx->out;
      taosMemoryFree(pOut->pCode);
      taosMemoryFree(pOut->pComment);
      taosMemoryFreeClear(pCtx->out);
      break;
    }
    case TDMT_MND_GET_USER_AUTH: {
      SGetUserAuthRsp* pOut = (SGetUserAuthRsp*)pCtx->out;
      taosHashCleanup(pOut->createdDbs);
      taosHashCleanup(pOut->readDbs);
      taosHashCleanup(pOut->writeDbs);
      taosMemoryFreeClear(pCtx->out);
      break;
    }
    default:
      qError("invalid reqType %d", pCtx->reqType);
      break;
  }
}

void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput) {
  if (NULL == pOutput) {
    return;
  }
  
  taosMemoryFree(pOutput->tbMeta);
  taosMemoryFree(pOutput);
}


void ctgResetTbMetaTask(SCtgTask* pTask) {
  SCtgTbMetaCtx* taskCtx = (SCtgTbMetaCtx*)pTask->taskCtx;
  memset(&taskCtx->tbInfo, 0, sizeof(taskCtx->tbInfo));
  taskCtx->flag = CTG_FLAG_UNKNOWN_STB;
  
  if (pTask->msgCtx.lastOut) {
    ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.lastOut);
    pTask->msgCtx.lastOut = NULL;
  }
  if (pTask->msgCtx.out) {
    ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.out);
    pTask->msgCtx.out = NULL;
  }
  taosMemoryFreeClear(pTask->msgCtx.target);
  taosMemoryFreeClear(pTask->res);
}

D
dapan1121 已提交
445 446
void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
  switch (type) {
D
dapan1121 已提交
447 448
    case CTG_TASK_GET_QNODE:
    case CTG_TASK_GET_DNODE:
D
dapan1121 已提交
449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466
    case CTG_TASK_GET_DB_VGROUP: {
      taosArrayDestroy((SArray*)*pRes);
      *pRes = NULL;
      break;
    }
    case CTG_TASK_GET_DB_CFG: {
      if (*pRes) {
        SDbCfgInfo* pInfo = (SDbCfgInfo*)*pRes;
        taosArrayDestroy(pInfo->pRetensions);
        taosMemoryFreeClear(*pRes);
      }
      break;
    }
    case CTG_TASK_GET_TB_INDEX: {
      taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
      *pRes = NULL;
      break;
    }
D
dapan1121 已提交
467 468 469 470 471 472 473 474
    case CTG_TASK_GET_TB_CFG: {
      if (*pRes) {
        STableCfg* pInfo = (STableCfg*)*pRes;
        tFreeSTableCfgRsp(pInfo);
        taosMemoryFreeClear(*pRes);
      }
      break;
    }
D
dapan1121 已提交
475 476 477 478 479 480 481
    case CTG_TASK_GET_TB_HASH:
    case CTG_TASK_GET_DB_INFO:
    case CTG_TASK_GET_INDEX:
    case CTG_TASK_GET_UDF: 
    case CTG_TASK_GET_USER: 
    case CTG_TASK_GET_SVR_VER:
    case CTG_TASK_GET_TB_META: {
D
dapan1121 已提交
482 483 484 485 486 487 488 489 490 491 492 493
      taosMemoryFreeClear(*pRes);
      break;
    }
    default:
      qError("invalid task type %d", type);
      break;
  }
}


void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void **pRes) {
  switch (type) {
D
dapan1121 已提交
494
    case CTG_TASK_GET_QNODE:
D
dapan1121 已提交
495 496 497 498 499
    case CTG_TASK_GET_DNODE: {
      taosArrayDestroy((SArray*)*pRes);
      *pRes = NULL;
      break;
    }
D
dapan1121 已提交
500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520
    case CTG_TASK_GET_DB_VGROUP: {
      if (*pRes) {
        SDBVgInfo* pInfo = (SDBVgInfo*)*pRes;
        taosHashCleanup(pInfo->vgHash);
        taosMemoryFreeClear(*pRes);
      }
      break;
    }
    case CTG_TASK_GET_DB_CFG: {
      if (*pRes) {
        SDbCfgInfo* pInfo = (SDbCfgInfo*)*pRes;
        taosArrayDestroy(pInfo->pRetensions);
        taosMemoryFreeClear(*pRes);
      }
      break;
    }
    case CTG_TASK_GET_TB_INDEX: {
      taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
      *pRes = NULL;
      break;
    }
D
dapan1121 已提交
521 522 523 524 525 526 527 528
    case CTG_TASK_GET_TB_CFG: {
      if (*pRes) {
        STableCfg* pInfo = (STableCfg*)*pRes;
        tFreeSTableCfgRsp(pInfo);
        taosMemoryFreeClear(*pRes);
      }
      break;
    }
D
dapan1121 已提交
529 530 531 532 533 534
    case CTG_TASK_GET_TB_META:    
    case CTG_TASK_GET_DB_INFO:
    case CTG_TASK_GET_TB_HASH:
    case CTG_TASK_GET_INDEX: 
    case CTG_TASK_GET_UDF: 
    case CTG_TASK_GET_SVR_VER: 
D
dapan1121 已提交
535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556
    case CTG_TASK_GET_USER: {
      taosMemoryFreeClear(*pRes);
      break;
    }
    default:
      qError("invalid task type %d", type);
      break;
  }
}


void ctgClearSubTaskRes(SCtgSubRes *pRes) {
  pRes->code = 0;

  if (NULL == pRes->res) {
    return;
  }

  ctgFreeSubTaskRes(pRes->type, &pRes->res);
}

void ctgFreeTaskCtx(SCtgTask* pTask) {
D
dapan1121 已提交
557 558 559 560 561 562 563 564
  switch (pTask->type) {
    case CTG_TASK_GET_TB_META: {
      SCtgTbMetaCtx* taskCtx = (SCtgTbMetaCtx*)pTask->taskCtx;
      taosMemoryFreeClear(taskCtx->pName);
      if (pTask->msgCtx.lastOut) {
        ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.lastOut);
        pTask->msgCtx.lastOut = NULL;
      }
D
dapan1121 已提交
565
      taosMemoryFreeClear(pTask->taskCtx);
D
dapan1121 已提交
566 567 568 569 570
      break;
    }
    case CTG_TASK_GET_TB_HASH: {
      SCtgTbHashCtx* taskCtx = (SCtgTbHashCtx*)pTask->taskCtx;
      taosMemoryFreeClear(taskCtx->pName);
D
dapan1121 已提交
571
      taosMemoryFreeClear(pTask->taskCtx);      
D
dapan1121 已提交
572 573
      break;
    }
D
dapan1121 已提交
574 575 576 577
    case CTG_TASK_GET_TB_INDEX: {
      SCtgTbIndexCtx* taskCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
      taosMemoryFreeClear(taskCtx->pName);
      taosMemoryFreeClear(pTask->taskCtx);
D
dapan1121 已提交
578 579 580 581 582 583 584
      break;
    }
    case CTG_TASK_GET_TB_CFG: {
      SCtgTbCfgCtx* taskCtx = (SCtgTbCfgCtx*)pTask->taskCtx;
      taosMemoryFreeClear(taskCtx->pName);
      taosMemoryFreeClear(taskCtx->pVgInfo);
      taosMemoryFreeClear(pTask->taskCtx);
D
dapan1121 已提交
585 586
      break;
    }
D
dapan1121 已提交
587 588 589 590 591 592
    case CTG_TASK_GET_DB_VGROUP:
    case CTG_TASK_GET_DB_CFG:
    case CTG_TASK_GET_DB_INFO:    
    case CTG_TASK_GET_INDEX:
    case CTG_TASK_GET_UDF:
    case CTG_TASK_GET_QNODE:    
D
dapan1121 已提交
593 594 595 596 597 598 599 600 601 602
    case CTG_TASK_GET_USER: {
      taosMemoryFreeClear(pTask->taskCtx);
      break;
    }
    default:
      qError("invalid task type %d", pTask->type);
      break;
  }
}

D
dapan1121 已提交
603 604 605 606 607 608 609 610 611 612

void ctgFreeTask(SCtgTask* pTask) {
  ctgFreeMsgCtx(&pTask->msgCtx);
  ctgFreeTaskRes(pTask->type, &pTask->res);
  ctgFreeTaskCtx(pTask);

  taosArrayDestroy(pTask->pParents);
  ctgClearSubTaskRes(&pTask->subRes);
}

D
dapan1121 已提交
613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637
void ctgFreeTasks(SArray* pArray) {
  if (NULL == pArray) {
    return;
  }

  int32_t num = taosArrayGetSize(pArray);
  for (int32_t i = 0; i < num; ++i) {
    SCtgTask* pTask = taosArrayGet(pArray, i);
    ctgFreeTask(pTask);
  }

  taosArrayDestroy(pArray);
}

void ctgFreeJob(void* job) {
  if (NULL == job) {
    return;
  }
  
  SCtgJob* pJob = (SCtgJob*)job;

  int64_t rid = pJob->refId;
  uint64_t qid = pJob->queryId;

  ctgFreeTasks(pJob->pTasks);
D
dapan1121 已提交
638
  ctgFreeBatchs(pJob->pBatchs);
D
dapan1121 已提交
639 640 641 642 643

  ctgFreeSMetaData(&pJob->jobRes);

  taosMemoryFree(job);

D
dapan1121 已提交
644
  qDebug("QID:0x%" PRIx64 ", ctg job 0x%" PRIx64 " freed", qid, rid);
D
dapan1121 已提交
645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684
}

int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target) {
  ctgFreeMsgCtx(pCtx);

  pCtx->reqType = reqType;
  pCtx->out = out;
  if (target) {
    pCtx->target = strdup(target);
    if (NULL == pCtx->target) {
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
    }
  } else {
    pCtx->target = NULL;
  }

  return TSDB_CODE_SUCCESS;
}


int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
  switch (hashMethod) {
    default:
      *fp = MurmurHash3_32;
      break;
  }

  return TSDB_CODE_SUCCESS;
}

int32_t ctgGenerateVgList(SCatalog *pCtg, SHashObj *vgHash, SArray** pList) {
  SHashObj *vgroupHash = NULL;
  SVgroupInfo *vgInfo = NULL;
  SArray *vgList = NULL;
  int32_t code = 0;
  int32_t vgNum = taosHashGetSize(vgHash);

  vgList = taosArrayInit(vgNum, sizeof(SVgroupInfo));
  if (NULL == vgList) {
    ctgError("taosArrayInit failed, num:%d", vgNum);
D
dapan1121 已提交
685
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);    
D
dapan1121 已提交
686 687 688 689 690 691 692 693 694
  }

  void *pIter = taosHashIterate(vgHash, NULL);
  while (pIter) {
    vgInfo = pIter;

    if (NULL == taosArrayPush(vgList, vgInfo)) {
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
      taosHashCancelIterate(vgHash, pIter);      
D
dapan1121 已提交
695
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758
    }
    
    pIter = taosHashIterate(vgHash, pIter);
    vgInfo = NULL;
  }

  *pList = vgList;

  ctgDebug("Got vgList from cache, vgNum:%d", vgNum);

  return TSDB_CODE_SUCCESS;

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
}


int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup) {
  int32_t code = 0;
  
  int32_t vgNum = taosHashGetSize(dbInfo->vgHash);
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);

  if (vgNum <= 0) {
    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", db, vgNum);
    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
  }

  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;

  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));

  char tbFullName[TSDB_TABLE_FNAME_LEN];
  tNameExtractFullName(pTableName, tbFullName);

  uint32_t hashValue = (*fp)(tbFullName, (uint32_t)strlen(tbFullName));

  void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
  while (pIter) {
    vgInfo = pIter;
    if (hashValue >= vgInfo->hashBegin && hashValue <= vgInfo->hashEnd) {
      taosHashCancelIterate(dbInfo->vgHash, pIter);
      break;
    }
    
    pIter = taosHashIterate(dbInfo->vgHash, pIter);
    vgInfo = NULL;
  }

  if (NULL == vgInfo) {
    ctgError("no hash range found for hash value [%u], db:%s, numOfVgId:%d", hashValue, db, taosHashGetSize(dbInfo->vgHash));
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  *pVgroup = *vgInfo;

S
Shengliang Guan 已提交
759
  ctgDebug("Got tb %s hash vgroup, vgId:%d, epNum %d, current %s port %d", tbFullName, vgInfo->vgId, vgInfo->epSet.numOfEps,
D
dapan1121 已提交
760 761
    vgInfo->epSet.eps[vgInfo->epSet.inUse].fqdn, vgInfo->epSet.eps[vgInfo->epSet.inUse].port);

D
dapan1121 已提交
762 763 764 765
  CTG_RET(code);
}

int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) {
D
dapan1121 已提交
766
  if (*(uint64_t *)key1 < ((SSTableVersion*)key2)->suid) {
D
dapan1121 已提交
767
    return -1;
D
dapan1121 已提交
768
  } else if (*(uint64_t *)key1 > ((SSTableVersion*)key2)->suid) {
D
dapan1121 已提交
769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785
    return 1;
  } else {
    return 0;
  }
}

int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2) {
  if (*(int64_t *)key1 < ((SDbVgVersion*)key2)->dbId) {
    return -1;
  } else if (*(int64_t *)key1 > ((SDbVgVersion*)key2)->dbId) {
    return 1;
  } else {
    return 0;
  }
}

int32_t ctgStbVersionSortCompare(const void* key1, const void* key2) {
D
dapan1121 已提交
786
  if (((SSTableVersion*)key1)->suid < ((SSTableVersion*)key2)->suid) {
D
dapan1121 已提交
787
    return -1;
D
dapan1121 已提交
788
  } else if (((SSTableVersion*)key1)->suid > ((SSTableVersion*)key2)->suid) {
D
dapan1121 已提交
789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811
    return 1;
  } else {
    return 0;
  }
}

int32_t ctgDbVgVersionSortCompare(const void* key1, const void* key2) {
  if (((SDbVgVersion*)key1)->dbId < ((SDbVgVersion*)key2)->dbId) {
    return -1;
  } else if (((SDbVgVersion*)key1)->dbId > ((SDbVgVersion*)key2)->dbId) {
    return 1;
  } else {
    return 0;
  }
}




int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst) {
  *dst = taosMemoryMalloc(sizeof(SDBVgInfo));
  if (NULL == *dst) {
    qError("malloc %d failed", (int32_t)sizeof(SDBVgInfo));
D
dapan1121 已提交
812
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
813 814 815 816 817 818 819 820 821
  }

  memcpy(*dst, src, sizeof(SDBVgInfo));

  size_t hashSize = taosHashGetSize(src->vgHash);
  (*dst)->vgHash = taosHashInit(hashSize, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
  if (NULL == (*dst)->vgHash) {
    qError("taosHashInit %d failed", (int32_t)hashSize);
    taosMemoryFreeClear(*dst);
D
dapan1121 已提交
822
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
823 824 825 826 827 828 829 830 831 832 833 834
  }

  int32_t *vgId = NULL;
  void *pIter = taosHashIterate(src->vgHash, NULL);
  while (pIter) {
    vgId = taosHashGetKey(pIter, NULL);

    if (taosHashPut((*dst)->vgHash, (void *)vgId, sizeof(int32_t), pIter, sizeof(SVgroupInfo))) {
      qError("taosHashPut failed, hashSize:%d", (int32_t)hashSize);
      taosHashCancelIterate(src->vgHash, pIter);
      taosHashCleanup((*dst)->vgHash);
      taosMemoryFreeClear(*dst);
D
dapan1121 已提交
835
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
836 837 838 839 840 841 842 843 844 845 846 847 848 849 850
    }
    
    pIter = taosHashIterate(src->vgHash, pIter);
  }


  return TSDB_CODE_SUCCESS;
}



int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput) {
  *pOutput = taosMemoryMalloc(sizeof(STableMetaOutput));
  if (NULL == *pOutput) {
    qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
D
dapan1121 已提交
851
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
852 853 854 855 856 857 858
  }

  memcpy(*pOutput, output, sizeof(STableMetaOutput));

  if (output->tbMeta) {
    int32_t metaSize = CTG_META_SIZE(output->tbMeta);
    (*pOutput)->tbMeta = taosMemoryMalloc(metaSize);
D
dapan1121 已提交
859
    qDebug("tbMeta cloned, size:%d, p:%p", metaSize, (*pOutput)->tbMeta);
D
dapan1121 已提交
860 861 862
    if (NULL == (*pOutput)->tbMeta) {
      qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
      taosMemoryFreeClear(*pOutput);
D
dapan1121 已提交
863
      CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
D
dapan1121 已提交
864 865 866 867 868 869 870 871
    }

    memcpy((*pOutput)->tbMeta, output->tbMeta, metaSize);
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
872 873 874 875 876 877 878 879 880 881 882 883 884 885
int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes) {
  if (NULL == pIndex) {
    *pRes = NULL;
    return TSDB_CODE_SUCCESS;
  }

  int32_t num = taosArrayGetSize(pIndex);
  *pRes = taosArrayInit(num, sizeof(STableIndexInfo));
  if (NULL == *pRes) {
    CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
  }

  for (int32_t i = 0; i < num; ++i) {
    STableIndexInfo *pInfo = taosArrayGet(pIndex, i);
D
dapan1121 已提交
886 887
    pInfo = taosArrayPush(*pRes, pInfo);
    pInfo->expr = strdup(pInfo->expr);
D
dapan1121 已提交
888 889 890 891 892 893
  }

  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
894 895
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, char* dbFName, int32_t vgId) {
  if (msgType == TDMT_VND_TABLE_META || msgType == TDMT_VND_TABLE_CFG || msgType == TDMT_VND_BATCH_META) {
D
dapan1121 已提交
896
    pMsgSendInfo->target.type = TARGET_TYPE_VNODE;
D
dapan1121 已提交
897
    pMsgSendInfo->target.vgId = vgId;
D
dapan1121 已提交
898 899 900 901 902 903 904
    pMsgSendInfo->target.dbFName = strdup(dbFName);
  } else {
    pMsgSendInfo->target.type = TARGET_TYPE_MNODE;
  }

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
905 906