Merge pull request #9117 from taosdata/catalog_dev

Catalog dev

include/common/taosmsg.h

@@ -841,7 +841,7 @@ typedef struct {
   int32_t    tversion;
   uint64_t   tuid;
   uint64_t   suid;
-  SVgroupMsg vgroup;
+  int32_t    vgId;
   SSchema    pSchema[];
 } STableMetaMsg;
 
@@ -869,6 +869,7 @@ typedef struct {
   int32_t     dbVgroupVersion;
   int32_t     dbVgroupNum;
   int32_t     dbHashRange;
+  int32_t     dbHashType;
   SVgroupInfo vgroupInfo[];
 //int32_t     vgIdList[];
 } SUseDbRspMsg;

include/libs/catalog/catalog.h

@@ -45,42 +45,10 @@ typedef struct SMetaData {
   SEpSet    *pEpSet;      // qnode epset list
 } SMetaData;
 
-typedef struct STableComInfo {
-  uint8_t numOfTags;      // the number of tags in schema
-  uint8_t precision;      // the number of precision
-  int16_t numOfColumns;   // the number of columns
-  int32_t rowSize;        // row size of the schema
-} STableComInfo;
-
-/*
- * ASSERT(sizeof(SCTableMeta) == 24)
- * ASSERT(tableType == TSDB_CHILD_TABLE)
- * The cached child table meta info. For each child table, 24 bytes are required to keep the essential table info.
- */
-typedef struct SCTableMeta {
-  int32_t  vgId:24;
-  int8_t   tableType;
-  uint64_t uid;
-  uint64_t suid;
-} SCTableMeta;
-
-/*
- * Note that the first 24 bytes of STableMeta are identical to SCTableMeta, it is safe to cast a STableMeta to be a SCTableMeta.
- */
-typedef struct STableMeta {
-  int32_t        vgId:24;
-  int8_t         tableType;
-  uint64_t       uid;
-  uint64_t       suid;
-  // if the table is TSDB_CHILD_TABLE, the following information is acquired from the corresponding super table meta info
-  int16_t        sversion;
-  int16_t        tversion;
-  STableComInfo  tableInfo;
-  SSchema        schema[];
-} STableMeta;
-
 typedef struct SCatalogCfg {
-
+  bool     enableVgroupCache;
+  uint32_t maxTblCacheNum;
+  uint32_t maxDBCacheNum;
 } SCatalogCfg;
 
 int32_t catalogInit(SCatalogCfg *cfg);
@@ -96,19 +64,25 @@ int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle)
 
 
 int32_t catalogGetVgroupVersion(struct SCatalog* pCatalog, int32_t* version);
-int32_t catalogGetVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray** pVgroupList);
-int32_t catalogUpdateVgroup(struct SCatalog* pCatalog, SVgroupListInfo* pVgroup);
+
+/**
+ * get cluster vgroup list.
+ * @pVgroupList  - hash of vgroup list, key:vgId, value:SVgroupInfo
+ * @return
+ */
+int32_t catalogGetVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SHashObj** pVgroupHash);
+int32_t catalogUpdateVgroupCache(struct SCatalog* pCatalog, SVgroupListInfo* pVgroup);
 
 
 
 int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
 int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, int32_t forceUpdate, SDBVgroupInfo** dbInfo);
-int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo);
+int32_t catalogUpdateDBVgroupCache(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo);
 
 
-int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pTableName, STableMeta* pTableMeta);
-int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const STableMeta* pTableMeta);
-int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const STableMeta* pTableMeta, STableMeta* pNewTableMeta);
+int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta** pTableMeta);
+int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName);
+int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pDBName, const char* pTableName, STableMeta* pTableMeta);
 
 
 /**

include/libs/query/query.h

@@ -24,10 +24,49 @@ extern "C" {
 
 typedef SVgroupListRspMsg SVgroupListInfo;
 
+typedef struct STableComInfo {
+  uint8_t numOfTags;      // the number of tags in schema
+  uint8_t precision;      // the number of precision
+  int16_t numOfColumns;   // the number of columns
+  int32_t rowSize;        // row size of the schema
+} STableComInfo;
+
+/*
+ * ASSERT(sizeof(SCTableMeta) == 24)
+ * ASSERT(tableType == TSDB_CHILD_TABLE)
+ * The cached child table meta info. For each child table, 24 bytes are required to keep the essential table info.
+ */
+typedef struct SCTableMeta {
+  int32_t  vgId:24;
+  int8_t   tableType;
+  uint64_t uid;
+  uint64_t suid;
+} SCTableMeta;
+
+/*
+ * Note that the first 24 bytes of STableMeta are identical to SCTableMeta, it is safe to cast a STableMeta to be a SCTableMeta.
+ */
+typedef struct STableMeta {
+  //BEGIN: KEEP THIS PART SAME WITH SCTableMeta
+  int32_t        vgId:24;
+  int8_t         tableType;
+  uint64_t       uid;
+  uint64_t       suid;
+  //END: KEEP THIS PART SAME WITH SCTableMeta
+  
+  // if the table is TSDB_CHILD_TABLE, the following information is acquired from the corresponding super table meta info
+  int16_t        sversion;
+  int16_t        tversion;
+  STableComInfo  tableInfo;
+  SSchema        schema[];
+} STableMeta;
+
+
 typedef struct SDBVgroupInfo {
   int32_t vgroupVersion;
   SArray *vgId;
   int32_t hashRange;
+  int32_t hashType;
 } SDBVgroupInfo;
 
 typedef struct SUseDbOutput {
@@ -36,6 +75,13 @@ typedef struct SUseDbOutput {
   SDBVgroupInfo *dbVgroup;
 } SUseDbOutput;
 
+typedef struct STableMetaOutput {
+  int32_t     metaNum;
+  char        ctbFname[TSDB_TABLE_FNAME_LEN];
+  char        tbFname[TSDB_TABLE_FNAME_LEN];  
+  SCTableMeta ctbMeta;
+  STableMeta *tbMeta;
+} STableMetaOutput;
 
 extern int32_t (*queryBuildMsg[TSDB_MSG_TYPE_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen);
 extern int32_t (*queryProcessMsgRsp[TSDB_MSG_TYPE_MAX])(void* output, char *msg, int32_t msgSize);

source/libs/catalog/inc/catalogInt.h

@@ -24,16 +24,16 @@ extern "C" {
 #include "common.h"
 #include "tlog.h"
 
-#define CTG_DEFAULT_CLUSTER_NUMBER 6
-#define CTG_DEFAULT_VGROUP_NUMBER 100
-#define CTG_DEFAULT_DB_NUMBER 20
+#define CTG_DEFAULT_CACHE_CLUSTER_NUMBER 6
+#define CTG_DEFAULT_CACHE_VGROUP_NUMBER 100
+#define CTG_DEFAULT_CACHE_DB_NUMBER 20
+#define CTG_DEFAULT_CACHE_TABLEMETA_NUMBER 100000
 
 #define CTG_DEFAULT_INVALID_VERSION (-1)
 
 typedef struct SVgroupListCache {
   int32_t vgroupVersion;
-  SHashObj *cache;        // key:vgId, value:SVgroupInfo*
-  SArray   *arrayCache;   // SVgroupInfo
+  SHashObj *cache;        // key:vgId, value:SVgroupInfo
 } SVgroupListCache;
 
 typedef struct SDBVgroupCache {
@@ -41,20 +41,23 @@ typedef struct SDBVgroupCache {
 } SDBVgroupCache;
 
 typedef struct STableMetaCache {
-  SHashObj *cache;     //key:fulltablename, value:STableMeta
+  SHashObj *cache;           //key:fulltablename, value:STableMeta
+  SHashObj *stableCache;     //key:suid, value:STableMeta*
 } STableMetaCache;
 
 typedef struct SCatalog {
   SVgroupListCache vgroupCache;
-  SDBVgroupCache dbCache;
-  STableMetaCache tableCache;
+  SDBVgroupCache   dbCache;
+  STableMetaCache  tableCache;
 } SCatalog;
 
 typedef struct SCatalogMgmt {
   void       *pMsgSender;   // used to send messsage to mnode to fetch necessary metadata
   SHashObj   *pCluster;     // items cached for each cluster, the hash key is the cluster-id got from mgmt node
+  SCatalogCfg cfg;
 } SCatalogMgmt;
 
+typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
 
 extern int32_t ctgDebugFlag;
 

source/libs/parser/src/insertParser.c

@@ -186,7 +186,7 @@ static int32_t buildMetaReq(SInsertParseContext* pCxt, SToken* pStname, SCatalog
 static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
   SCatalogReq req;
   CHECK_CODE(buildMetaReq(pCxt, pTname, &req));
-  CHECK_CODE(catalogGetTableMeta(pCxt->pCatalog, NULL, NULL, NULL, &pCxt->meta)); //TODO
+  CHECK_CODE(catalogGetTableMeta(pCxt->pCatalog, NULL, NULL, NULL, NULL, NULL)); //TODO
   pCxt->pTableMeta = (STableMeta*)taosArrayGetP(pCxt->meta.pTableMeta, 0);
   return TSDB_CODE_SUCCESS;
 }

source/libs/parser/src/parserUtil.c

@@ -1464,29 +1464,6 @@ int32_t copyTagData(STagData* dst, const STagData* src) {
   return 0;
 }
 
-STableMeta* createSuperTableMeta(STableMetaMsg* pChild) {
-  assert(pChild != NULL);
-  int32_t total = pChild->numOfColumns + pChild->numOfTags;
-
-  STableMeta* pTableMeta = calloc(1, sizeof(STableMeta) + sizeof(SSchema) * total);
-  pTableMeta->tableType = TSDB_SUPER_TABLE;
-  pTableMeta->tableInfo.numOfTags = pChild->numOfTags;
-  pTableMeta->tableInfo.numOfColumns = pChild->numOfColumns;
-  pTableMeta->tableInfo.precision = pChild->precision;
-
-  pTableMeta->uid = pChild->suid;
-  pTableMeta->tversion = pChild->tversion;
-  pTableMeta->sversion = pChild->sversion;
-
-  memcpy(pTableMeta->schema, pChild->pSchema, sizeof(SSchema) * total);
-
-  int32_t num = pTableMeta->tableInfo.numOfColumns;
-  for(int32_t i = 0; i < num; ++i) {
-    pTableMeta->tableInfo.rowSize += pTableMeta->schema[i].bytes;
-  }
-
-  return pTableMeta;
-}
 
 uint32_t getTableMetaSize(const STableMeta* pTableMeta) {
   assert(pTableMeta != NULL);

source/libs/query/src/querymsg.c

@@ -222,6 +222,13 @@ int32_t queryProcessUseDBRsp(void* output, char *msg, int32_t msgSize) {
     
     pOut->dbVgroup->vgroupVersion = pRsp->dbVgroupVersion;
     pOut->dbVgroup->hashRange = htonl(pRsp->dbHashRange);
+    pOut->dbVgroup->hashType = htonl(pRsp->dbHashType);
+
+    if (pOut->dbVgroup->hashRange < 0) {
+      qError("invalid hashRange[%d] for db[%s]", pOut->dbVgroup->hashRange, pRsp->db);
+      code = TSDB_CODE_TSC_INVALID_INPUT;
+      goto _exit;
+    }
 
     for (int32_t i = 0; i < pRsp->dbVgroupNum; ++i) {
       *(vgIdList + i) = htonl(*(vgIdList + i));
@@ -244,13 +251,134 @@ _exit:
   return code;
 }
 
+static int32_t queryConvertTableMetaMsg(STableMetaMsg* pMetaMsg) {
+  pMetaMsg->numOfTags = htonl(pMetaMsg->numOfTags);
+  pMetaMsg->numOfColumns = htonl(pMetaMsg->numOfColumns);
+  pMetaMsg->sversion = htonl(pMetaMsg->sversion);
+  pMetaMsg->tversion = htonl(pMetaMsg->tversion);
+  pMetaMsg->tuid = htobe64(pMetaMsg->tuid);
+  pMetaMsg->suid = htobe64(pMetaMsg->suid);
+  pMetaMsg->vgId = htonl(pMetaMsg->vgId);
+
+  if (pMetaMsg->numOfTags < 0 || pMetaMsg->numOfTags > TSDB_MAX_TAGS) {
+    qError("invalid numOfTags[%d] in table meta rsp msg", pMetaMsg->numOfTags);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  if (pMetaMsg->numOfColumns > TSDB_MAX_COLUMNS || pMetaMsg->numOfColumns <= 0) {
+    qError("invalid numOfColumns[%d] in table meta rsp msg", pMetaMsg->numOfColumns);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  if (pMetaMsg->tableType != TSDB_SUPER_TABLE && pMetaMsg->tableType != TSDB_CHILD_TABLE && pMetaMsg->tableType != TSDB_NORMAL_TABLE) {
+    qError("invalid tableType[%d] in table meta rsp msg", pMetaMsg->tableType);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  if (pMetaMsg->sversion < 0) {
+    qError("invalid sversion[%d] in table meta rsp msg", pMetaMsg->sversion);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  if (pMetaMsg->tversion < 0) {
+    qError("invalid tversion[%d] in table meta rsp msg", pMetaMsg->tversion);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+  
+  SSchema* pSchema = pMetaMsg->pSchema;
+
+  int32_t numOfTotalCols = pMetaMsg->numOfColumns + pMetaMsg->numOfTags;
+  for (int i = 0; i < numOfTotalCols; ++i) {
+    pSchema->bytes = htonl(pSchema->bytes);
+    pSchema->colId = htonl(pSchema->colId);
+
+    pSchema++;
+  }
+
+  if (pMetaMsg->pSchema[0].colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
+    qError("invalid colId[%d] for the first column in table meta rsp msg", pMetaMsg->pSchema[0].colId);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t queryCreateTableMetaFromMsg(STableMetaMsg* msg, bool isSuperTable, STableMeta **pMeta) {
+  int32_t total = msg->numOfColumns + msg->numOfTags;
+  int32_t metaSize = sizeof(STableMeta) + sizeof(SSchema) * total;
+  
+  STableMeta* pTableMeta = calloc(1, metaSize);
+  if (NULL == pTableMeta) {
+    qError("calloc size[%d] failed", metaSize);
+    return TSDB_CODE_TSC_OUT_OF_MEMORY;
+  }
+  
+  pTableMeta->tableType = isSuperTable ? TSDB_SUPER_TABLE : msg->tableType;
+  pTableMeta->uid = msg->suid;
+  pTableMeta->suid = msg->suid;
+  pTableMeta->sversion = msg->sversion;
+  pTableMeta->tversion = msg->tversion;
+
+  pTableMeta->tableInfo.numOfTags = msg->numOfTags;
+  pTableMeta->tableInfo.precision = msg->precision;
+  pTableMeta->tableInfo.numOfColumns = msg->numOfColumns;
+
+  for(int32_t i = 0; i < msg->numOfColumns; ++i) {
+    pTableMeta->tableInfo.rowSize += pTableMeta->schema[i].bytes;
+  }
+
+  memcpy(pTableMeta->schema, msg->pSchema, sizeof(SSchema) * total);
+
+  *pMeta = pTableMeta;
+  
+  return TSDB_CODE_SUCCESS;
+}
+
+
+int32_t queryProcessTableMetaRsp(void* output, char *msg, int32_t msgSize) {
+  STableMetaMsg *pMetaMsg = (STableMetaMsg *)msg;
+  int32_t code = queryConvertTableMetaMsg(pMetaMsg);
+  if (code != TSDB_CODE_SUCCESS) {
+    return code;
+  }
+
+  STableMetaOutput *pOut = (STableMetaOutput *)output;
+  
+  if (!tIsValidSchema(pMetaMsg->pSchema, pMetaMsg->numOfColumns, pMetaMsg->numOfTags)) {
+    qError("validate table meta schema in rsp msg failed");
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  if (pMetaMsg->tableType == TSDB_CHILD_TABLE) {
+    pOut->metaNum = 2;
+    
+    memcpy(pOut->ctbFname, pMetaMsg->tbFname, sizeof(pOut->ctbFname));
+    memcpy(pOut->tbFname, pMetaMsg->stbFname, sizeof(pOut->tbFname));
+    
+    pOut->ctbMeta.vgId = pMetaMsg->vgId;
+    pOut->ctbMeta.tableType = pMetaMsg->tableType;
+    pOut->ctbMeta.uid = pMetaMsg->tuid;
+    pOut->ctbMeta.suid = pMetaMsg->suid;
+
+    code = queryCreateTableMetaFromMsg(pMetaMsg, true, &pOut->tbMeta);
+  } else {
+    pOut->metaNum = 1;
+    
+    memcpy(pOut->tbFname, pMetaMsg->tbFname, sizeof(pOut->tbFname));
+    
+    code = queryCreateTableMetaFromMsg(pMetaMsg, false, &pOut->tbMeta);
+  }
+  
+  return code;
+}
+
 
 void msgInit() {
   queryBuildMsg[TSDB_MSG_TYPE_TABLE_META] = queryBuildTableMetaReqMsg;
   queryBuildMsg[TSDB_MSG_TYPE_VGROUP_LIST] = queryBuildVgroupListReqMsg;
   queryBuildMsg[TSDB_MSG_TYPE_USE_DB] = queryBuildUseDbMsg;
 
-  //tscProcessMsgRsp[TSDB_MSG_TYPE_TABLE_META] = tscProcessTableMetaRsp;
+  queryProcessMsgRsp[TSDB_MSG_TYPE_TABLE_META] = queryProcessTableMetaRsp;
   queryProcessMsgRsp[TSDB_MSG_TYPE_VGROUP_LIST] = queryProcessVgroupListRsp;
   queryProcessMsgRsp[TSDB_MSG_TYPE_USE_DB] = queryProcessUseDBRsp;