From e36bf05f98b7b973f742df9ab7238a4cad518f3f Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Thu, 20 Apr 2023 17:30:09 +0800 Subject: [PATCH] enh: support get subtable tag and auth --- include/libs/catalog/catalog.h | 1 + source/libs/catalog/inc/catalogInt.h | 10 +++ source/libs/catalog/src/catalog.c | 2 +- source/libs/catalog/src/ctgAsync.c | 96 +++++++++++++++++---- source/libs/catalog/src/ctgCache.c | 23 ++++++ source/libs/catalog/src/ctgUtil.c | 119 ++++++++++++++++++--------- 6 files changed, 196 insertions(+), 55 deletions(-) diff --git a/include/libs/catalog/catalog.h b/include/libs/catalog/catalog.h index 2f8e7846f3..6f2fb4eb6b 100644 --- a/include/libs/catalog/catalog.h +++ b/include/libs/catalog/catalog.h @@ -106,6 +106,7 @@ typedef struct SMetaData { SArray* pUser; // pRes = SUserAuthRes* SArray* pQnodeList; // pRes = SArray* SArray* pTableCfg; // pRes = STableCfg* + SArray* pTableTag; // pRes = SArray* SArray* pDnodeList; // pRes = SArray* SMetaRes* pSvrVer; // pRes = char* } SMetaData; diff --git a/source/libs/catalog/inc/catalogInt.h b/source/libs/catalog/inc/catalogInt.h index f0e5024c59..1eaf45dafe 100644 --- a/source/libs/catalog/inc/catalogInt.h +++ b/source/libs/catalog/inc/catalogInt.h @@ -154,6 +154,11 @@ typedef struct SCtgTbCacheInfo { int32_t tbType; } SCtgTbCacheInfo; +typedef struct SCtgTbMetaParam { + SName* pName; + int32_t flag; +} SCtgTbMetaParam; + typedef struct SCtgTbMetaCtx { SCtgTbCacheInfo tbInfo; int32_t vgId; @@ -631,6 +636,7 @@ typedef struct SCtgCacheItemInfo { #define CTG_FLAG_SYS_DB 0x8 #define CTG_FLAG_FORCE_UPDATE 0x10 #define CTG_FLAG_ONLY_CACHE 0x20 +#define CTG_FLAG_SYNC_OP 0x40 #define CTG_FLAG_SET(_flag, _v) ((_flag) |= (_v)) @@ -933,6 +939,10 @@ void ctgReleaseVgMetaToCache(SCatalog* pCtg, SCtgDBCache* dbCache, SCtgTbCach void ctgReleaseTbMetaToCache(SCatalog* pCtg, SCtgDBCache* dbCache, SCtgTbCache* pCache); void ctgGetGlobalCacheStat(SCtgCacheStat* pStat); int32_t ctgChkSetAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res); +int32_t ctgGetTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta); +int32_t ctgGetCachedStbNameFromSuid(SCatalog* pCtg, char* dbFName, uint64_t suid, char **stbName); +int32_t ctgGetTbTagCb(SCtgTask* pTask); +int32_t ctgGetUserCb(SCtgTask* pTask); extern SCatalogMgmt gCtgMgmt; extern SCtgDebug gCTGDebug; diff --git a/source/libs/catalog/src/catalog.c b/source/libs/catalog/src/catalog.c index bddc6c01a7..b263654e70 100644 --- a/source/libs/catalog/src/catalog.c +++ b/source/libs/catalog/src/catalog.c @@ -208,7 +208,7 @@ int32_t ctgGetTbMeta(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetaCtx* ctx } while (true) { - CTG_ERR_JRET(ctgRefreshTbMeta(pCtg, pConn, ctx, &output, false)); + CTG_ERR_JRET(ctgRefreshTbMeta(pCtg, pConn, ctx, &output, ctx->flag & CTG_FLAG_SYNC_OP)); if (CTG_IS_META_TABLE(output->metaType)) { *pTableMeta = output->tbMeta; diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c index b10da5cc73..affcfeb8ac 100644 --- a/source/libs/catalog/src/ctgAsync.c +++ b/source/libs/catalog/src/ctgAsync.c @@ -21,7 +21,8 @@ #include "trpc.h" int32_t ctgInitGetTbMetaTask(SCtgJob* pJob, int32_t taskIdx, void* param) { - SName* name = (SName*)param; + SCtgTbMetaParam* pParam = (SCtgTbMetaParam*)param; + SName* name = pParam->pName; SCtgTask task = {0}; task.type = CTG_TASK_GET_TB_META; @@ -41,7 +42,7 @@ int32_t ctgInitGetTbMetaTask(SCtgJob* pJob, int32_t taskIdx, void* param) { } memcpy(ctx->pName, name, sizeof(*name)); - ctx->flag = CTG_FLAG_UNKNOWN_STB; + ctx->flag = pParam->flag | CTG_FLAG_UNKNOWN_STB; taosArrayPush(pJob->pTasks, &task); @@ -545,7 +546,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const int32_t dbInfoNum = (int32_t)taosArrayGetSize(pReq->pDbInfo); int32_t tbIndexNum = (int32_t)taosArrayGetSize(pReq->pTableIndex); int32_t tbCfgNum = (int32_t)taosArrayGetSize(pReq->pTableCfg); - int32_t tbTagNum = (int32_t)ctgGetTablesReqNum(pReq->pTableTag); + int32_t tbTagNum = (int32_t)taosArrayGetSize(pReq->pTableTag); int32_t taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dnodeNum + svrVerNum + dbCfgNum + indexNum + userNum + dbInfoNum + tbIndexNum + tbCfgNum + tbTagNum; @@ -646,7 +647,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_TB_CFG, name, NULL)); } - for (int32_t i = 0; i < tbCfgNum; ++i) { + for (int32_t i = 0; i < tbTagNum; ++i) { SName* name = taosArrayGet(pReq->pTableTag, i); CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_TB_TAG, name, NULL)); } @@ -761,7 +762,11 @@ int32_t ctgDumpTbHashsRes(SCtgTask* pTask) { int32_t ctgDumpTbIndexRes(SCtgTask* pTask) { SCtgJob* pJob = pTask->pJob; if (NULL == pJob->jobRes.pTableIndex) { - pJob->jobRes.pTableIndex = taosArrayInit(pJob->tbIndexNum, sizeof(SMetaRes)); + SArray* pRes = taosArrayInit(pJob->tbIndexNum, sizeof(SMetaRes)); + if (atomic_val_compare_exchange_ptr(&pJob->jobRes.pTableIndex, NULL, pRes)) { + taosArrayDestroy(pRes); + } + if (NULL == pJob->jobRes.pTableIndex) { CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); } @@ -776,7 +781,11 @@ int32_t ctgDumpTbIndexRes(SCtgTask* pTask) { int32_t ctgDumpTbCfgRes(SCtgTask* pTask) { SCtgJob* pJob = pTask->pJob; if (NULL == pJob->jobRes.pTableCfg) { - pJob->jobRes.pTableCfg = taosArrayInit(pJob->tbCfgNum, sizeof(SMetaRes)); + SArray* pRes = taosArrayInit(pJob->tbCfgNum, sizeof(SMetaRes)); + if (atomic_val_compare_exchange_ptr(&pJob->jobRes.pTableCfg, NULL, pRes)) { + taosArrayDestroy(pRes); + } + if (NULL == pJob->jobRes.pTableCfg) { CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); } @@ -788,6 +797,26 @@ int32_t ctgDumpTbCfgRes(SCtgTask* pTask) { return TSDB_CODE_SUCCESS; } +int32_t ctgDumpTbTagRes(SCtgTask* pTask) { + SCtgJob* pJob = pTask->pJob; + if (NULL == pJob->jobRes.pTableTag) { + SArray* pRes = taosArrayInit(pJob->tbTagNum, sizeof(SMetaRes)); + if (atomic_val_compare_exchange_ptr(&pJob->jobRes.pTableTag, NULL, pRes)) { + taosArrayDestroy(pRes); + } + + if (NULL == pJob->jobRes.pTableTag) { + CTG_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); + } + } + + SMetaRes res = {.code = pTask->code, .pRes = pTask->res}; + taosArrayPush(pJob->jobRes.pTableTag, &res); + + return TSDB_CODE_SUCCESS; +} + + int32_t ctgDumpIndexRes(SCtgTask* pTask) { SCtgJob* pJob = pTask->pJob; if (NULL == pJob->jobRes.pIndex) { @@ -1123,7 +1152,7 @@ int32_t ctgHandleGetTbMetaRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBuf STableMetaOutput* pOut = (STableMetaOutput*)pMsgCtx->out; - ctgUpdateTbMetaToCache(pCtg, pOut, false); + ctgUpdateTbMetaToCache(pCtg, pOut, flag & CTG_FLAG_SYNC_OP); if (CTG_IS_META_BOTH(pOut->metaType)) { memcpy(pOut->tbMeta, &pOut->ctbMeta, sizeof(pOut->ctbMeta)); @@ -1525,11 +1554,36 @@ _return: int32_t ctgHandleGetTbTagRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBuf* pMsg, int32_t rspCode) { int32_t code = 0; SCtgTask* pTask = tReq->pTask; + SCatalog* pCtg = pTask->pJob->pCtg; CTG_ERR_JRET(ctgProcessRspMsg(&pTask->msgCtx.out, reqType, pMsg->pData, pMsg->len, rspCode, pTask->msgCtx.target)); STableCfgRsp* pRsp = (STableCfgRsp*)pTask->msgCtx.out; + if (NULL == pRsp->pTags || pRsp->tagsLen <= 0) { + ctgError("invalid tag in tbCfg rsp, pTags:%p, len:%d", pRsp->pTags, pRsp->tagsLen); + CTG_ERR_JRET(TSDB_CODE_INVALID_MSG); + } + + SArray* pTagVals = NULL; + STag* pTag = (STag*)pRsp->pTags; + + if (tTagIsJson(pTag)) { + pTagVals = taosArrayInit(1, sizeof(STagVal)); + if (NULL == pTagVals) { + CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY); + } + + char* pJson = parseTagDatatoJson(pTag); + STagVal tagVal; + tagVal.cid = 0; + tagVal.type = TSDB_DATA_TYPE_JSON; + tagVal.pData = pJson; + tagVal.nData = strlen(pJson); + taosArrayPush(pTagVals, &tagVal); + } else { + CTG_ERR_JRET(tTagToValArray((const STag*)pRsp->pTags, &pTagVals)); + } - TSWAP(pTask->res, pTask->msgCtx.out); + pTask->res = pTagVals; _return: @@ -1971,7 +2025,10 @@ int32_t ctgLaunchGetTbCfgTask(SCtgTask* pTask) { if (pCtx->tbType <= 0) { CTG_ERR_JRET(ctgReadTbTypeFromCache(pCtg, dbFName, pCtx->pName->tname, &pCtx->tbType)); if (pCtx->tbType <= 0) { - CTG_ERR_JRET(ctgLaunchSubTask(pTask, CTG_TASK_GET_TB_META, ctgGetTbCfgCb, pCtx->pName)); + SCtgTbMetaParam param; + param.pName = pCtx->pName; + param.flag = 0; + CTG_ERR_JRET(ctgLaunchSubTask(pTask, CTG_TASK_GET_TB_META, ctgGetTbCfgCb, ¶m)); return TSDB_CODE_SUCCESS; } } @@ -2019,7 +2076,7 @@ int32_t ctgLaunchGetTbTagTask(SCtgTask* pTask) { if (NULL == pCtx->pVgInfo) { CTG_ERR_JRET(ctgGetTbHashVgroupFromCache(pCtg, pCtx->pName, &pCtx->pVgInfo)); if (NULL == pCtx->pVgInfo) { - CTG_ERR_JRET(ctgLaunchSubTask(pTask, CTG_TASK_GET_DB_VGROUP, ctgGetTbCfgCb, dbFName)); + CTG_ERR_JRET(ctgLaunchSubTask(pTask, CTG_TASK_GET_DB_VGROUP, ctgGetTbTagCb, dbFName)); return TSDB_CODE_SUCCESS; } } @@ -2189,7 +2246,10 @@ int32_t ctgLaunchGetUserTask(SCtgTask* pTask) { taosMemoryFreeClear(rsp.pRawRes); if (rsp.metaNotExists) { - CTG_ERR_RET(ctgLaunchSubTask(pTask, CTG_TASK_GET_TB_META, ctgGetTbCfgCb, &pCtx->user.tbName)); + SCtgTbMetaParam param; + param.pName = &pCtx->user.tbName; + param.flag = CTG_FLAG_SYNC_OP; + CTG_ERR_RET(ctgLaunchSubTask(pTask, CTG_TASK_GET_TB_META, ctgGetUserCb, ¶m)); } else { CTG_ERR_RET(ctgGetUserDbAuthFromMnode(pCtg, pConn, pCtx->user.user, NULL, pTask)); } @@ -2251,9 +2311,11 @@ int32_t ctgGetTbTagCb(SCtgTask* pTask) { SCtgTbTagCtx* pCtx = (SCtgTbTagCtx*)pTask->taskCtx; SDBVgInfo* pDb = (SDBVgInfo*)pTask->subRes.res; - pCtx->pVgInfo = taosMemoryCalloc(1, sizeof(SVgroupInfo)); - CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pTask->pJob->pCtg, pDb, pCtx->pName, pCtx->pVgInfo)); - + if (NULL == pCtx->pVgInfo) { + pCtx->pVgInfo = taosMemoryCalloc(1, sizeof(SVgroupInfo)); + CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pTask->pJob->pCtg, pDb, pCtx->pName, pCtx->pVgInfo)); + } + CTG_RET(ctgLaunchGetTbTagTask(pTask)); _return: @@ -2286,8 +2348,12 @@ int32_t ctgCompDbVgTasks(SCtgTask* pTask, void* param, bool* equal) { int32_t ctgCompTbMetaTasks(SCtgTask* pTask, void* param, bool* equal) { SCtgTbMetaCtx* ctx = pTask->taskCtx; + SCtgTbMetaParam* pParam = (SCtgTbMetaParam*)param; - *equal = tNameTbNameEqual(ctx->pName, (SName*)param); + *equal = tNameTbNameEqual(ctx->pName, (SName*)pParam->pName); + if (*equal) { + ctx->flag |= pParam->flag; + } return TSDB_CODE_SUCCESS; } diff --git a/source/libs/catalog/src/ctgCache.c b/source/libs/catalog/src/ctgCache.c index 592b6e9c72..ee864d985e 100644 --- a/source/libs/catalog/src/ctgCache.c +++ b/source/libs/catalog/src/ctgCache.c @@ -703,6 +703,29 @@ _return: CTG_RET(code); } +int32_t ctgGetCachedStbNameFromSuid(SCatalog* pCtg, char* dbFName, uint64_t suid, char **stbName) { + *stbName = NULL; + + SCtgDBCache *dbCache = NULL; + ctgAcquireDBCache(pCtg, dbFName, &dbCache); + if (NULL == dbCache) { + ctgDebug("db %s not in cache", dbFName); + return TSDB_CODE_SUCCESS; + } + + char *stb = taosHashAcquire(dbCache->stbCache, &suid, sizeof(suid)); + if (NULL == stb) { + ctgDebug("stb 0x%" PRIx64 " not in cache, dbFName:%s", suid, dbFName); + return TSDB_CODE_SUCCESS; + } + + *stbName = taosStrdup(stb); + + taosHashRelease(dbCache->stbCache, stb); + + return TSDB_CODE_SUCCESS; +} + int32_t ctgChkAuthFromCache(SCatalog *pCtg, SUserAuthInfo *pReq, bool *inCache, SCtgAuthRsp *pRes) { if (IS_SYS_DBNAME(pReq->tbName.dbname)) { *inCache = true; diff --git a/source/libs/catalog/src/ctgUtil.c b/source/libs/catalog/src/ctgUtil.c index b2b2b5a87e..7ef2e34d1e 100644 --- a/source/libs/catalog/src/ctgUtil.c +++ b/source/libs/catalog/src/ctgUtil.c @@ -486,6 +486,18 @@ void ctgFreeBatchHash(void* hash) { taosMemoryFreeClear(pRes->pRes); } +void ctgFreeJsonTagVal(void *val) { + if (NULL == val) { + return; + } + + STagVal *pVal = (STagVal *)val; + + if (TSDB_DATA_TYPE_JSON == pVal->type) { + taosMemoryFree(pVal->pData); + } +} + void ctgFreeTaskRes(CTG_TASK_TYPE type, void** pRes) { switch (type) { case CTG_TASK_GET_QNODE: @@ -526,6 +538,14 @@ void ctgFreeTaskRes(CTG_TASK_TYPE type, void** pRes) { taosMemoryFreeClear(*pRes); break; } + case CTG_TASK_GET_TB_TAG: { + if (1 == taosArrayGetSize(*pRes)) { + taosArrayDestroyEx(*pRes, ctgFreeJsonTagVal); + } else { + taosArrayDestroy(*pRes); + } + *pRes = NULL; + } case CTG_TASK_GET_TB_META_BATCH: { SArray* pArray = (SArray*)*pRes; int32_t num = taosArrayGetSize(pArray); @@ -679,6 +699,12 @@ void ctgFreeTaskCtx(SCtgTask* pTask) { taosMemoryFreeClear(pTask->taskCtx); break; } + case CTG_TASK_GET_TB_TAG: { + SCtgTbTagCtx* taskCtx = (SCtgTbTagCtx*)pTask->taskCtx; + taosMemoryFreeClear(taskCtx->pName); + taosMemoryFreeClear(taskCtx->pVgInfo); + break; + } case CTG_TASK_GET_DB_VGROUP: case CTG_TASK_GET_DB_CFG: case CTG_TASK_GET_DB_INFO: @@ -1336,53 +1362,68 @@ int32_t ctgChkSetTbAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res) { STableMeta* pMeta = NULL; SGetUserAuthRsp* pInfo = &req->authInfo; SHashObj* pTbs = (AUTH_TYPE_READ == req->singleType) ? pInfo->readTbs : pInfo->writeTbs; + char* stbName = NULL; + + char tbFName[TSDB_TABLE_FNAME_LEN]; + char dbFName[TSDB_DB_FNAME_LEN]; + tNameExtractFullName(&req->pRawReq->tbName, tbFName); + tNameGetFullDbName(&req->pRawReq->tbName, dbFName); + + while (true) { + char* pCond = taosHashGet(pTbs, tbFName, strlen(tbFName)); + if (pCond) { + if (strlen(pCond) > 1) { + CTG_ERR_RET(nodesStringToNode(pCond, &res->pRawRes->pCond)); + } - char tbFullName[TSDB_TABLE_FNAME_LEN]; - tNameExtractFullName(&req->pRawReq->tbName, tbFullName); - char* pCond = taosHashGet(pTbs, tbFullName, strlen(tbFullName)); - if (pCond) { - if (strlen(pCond) > 1) { - CTG_ERR_RET(nodesStringToNode(pCond, &res->pRawRes->pCond)); + res->pRawRes->pass = true; + return TSDB_CODE_SUCCESS; } - res->pRawRes->pass = true; - return TSDB_CODE_SUCCESS; - } + if (stbName) { + res->pRawRes->pass = false; + goto _return; + } - res->pRawRes->pass = false; + CTG_ERR_RET(catalogGetCachedTableMeta(pCtg, &req->pRawReq->tbName, &pMeta)); + if (NULL == pMeta) { + if (req->onlyCache) { + res->metaNotExists = true; + ctgDebug("db %s tb %s meta not in cache for auth", req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname); + return TSDB_CODE_SUCCESS; + } - // CTG_ERR_RET(catalogGetCachedTableMeta(pCtg, &req->pRawReq->tbName, &pMeta)); - // if (NULL == pMeta) { - // if (req->onlyCache) { - // res->metaNotExists = true; - // ctgDebug("db %s tb %s meta not in cache for auth", req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname); - // return TSDB_CODE_SUCCESS; - // } + SCtgTbMetaCtx ctx = {0}; + ctx.pName = (SName*)&req->pRawReq->tbName; + ctx.flag = CTG_FLAG_UNKNOWN_STB | CTG_FLAG_SYNC_OP; + + CTG_ERR_RET(ctgGetTbMeta(pCtg, req->pConn, &ctx, &pMeta)); + } - // CTG_ERR_RET(catalogGetTableMeta(pCtg, req->pConn, &req->pRawReq->tbName, &pMeta)); - // } + if (TSDB_SUPER_TABLE == pMeta->tableType || TSDB_NORMAL_TABLE == pMeta->tableType) { + res->pRawRes->pass = false; + goto _return; + } - // if (TSDB_SUPER_TABLE == pMeta->tableType || TSDB_NORMAL_TABLE == pMeta->tableType) { - // res->pRawRes->pass = false; - // goto _return; - // } + if (TSDB_CHILD_TABLE == pMeta->tableType) { + CTG_ERR_JRET(ctgGetCachedStbNameFromSuid(pCtg, dbFName, pMeta->suid, &stbName)); + if (NULL == stbName) { + if (req->onlyCache) { + res->metaNotExists = true; + ctgDebug("suid %" PRIu64 " name not in cache for auth", pMeta->suid); + return TSDB_CODE_SUCCESS; + } - // if (TSDB_CHILD_TABLE == pMeta->tableType) { - // res->pRawRes->pass = true; + continue; + } - // /* - // char stbName[TSDB_TABLE_NAME_LEN] = {0}; - // CTG_ERR_JRET(ctgGetCachedStbNameFromSuid(pCtg, pMeta->suid, stbName)); - // if (0 == stbName[0]) { - // if (req->onlyCache) { - // res->notExists = true; - // return TSDB_CODE_SUCCESS; - // } + sprintf(tbFName, "%s.%s", dbFName, stbName); + continue; + } - // CTG_ERR_RET(catalogRefreshTableMeta(pCtg, req->pConn, &req->pRawReq->tbName, 0)); - // } - // */ - // } + ctgError("Invalid table type %d for %s", pMeta->tableType, tbFName); + CTG_ERR_JRET(TSDB_CODE_INVALID_PARA); + } _return: @@ -1423,7 +1464,7 @@ int32_t ctgChkSetAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res) { if (pInfo->readTbs && taosHashGetSize(pInfo->readTbs) > 0) { req->singleType = AUTH_TYPE_READ; CTG_ERR_RET(ctgChkSetTbAuthRes(pCtg, req, res)); - if (pRes->pass) { + if (pRes->pass || res->metaNotExists) { return TSDB_CODE_SUCCESS; } } @@ -1439,7 +1480,7 @@ int32_t ctgChkSetAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res) { if (pInfo->writeTbs && taosHashGetSize(pInfo->writeTbs) > 0) { req->singleType = AUTH_TYPE_WRITE; CTG_ERR_RET(ctgChkSetTbAuthRes(pCtg, req, res)); - if (pRes->pass) { + if (pRes->pass || res->metaNotExists) { return TSDB_CODE_SUCCESS; } } -- GitLab