enh: support get subtable tag and auth

include/libs/catalog/catalog.h

@@ -106,6 +106,7 @@ typedef struct SMetaData {
   SArray*   pUser;        // pRes = SUserAuthRes*
   SArray*   pQnodeList;   // pRes = SArray<SQueryNodeLoad>*
   SArray*   pTableCfg;    // pRes = STableCfg*
+  SArray*   pTableTag;    // pRes = SArray<STagVal>*
   SArray*   pDnodeList;   // pRes = SArray<SEpSet>*
   SMetaRes* pSvrVer;      // pRes = char*
 } SMetaData;

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;

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;

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, &param));
       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, &param));
   } 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;
 }

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;

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;
         }
       }