xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

catalog init version

This commit is contained in:
dapan1121 2021-12-11 11:18:15 +08:00
parent a9f9194a31
commit d962a2715f
9 changed files with 63 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
include/libs/catalog/catalog.h
View File

@ -30,19 +30,19 @@ extern "C" {
struct SCatalog; struct SCatalog;
typedef struct SMetaReq { typedef struct SCatalogReq {
char clusterId[TSDB_CLUSTER_ID_LEN]; char clusterId[TSDB_CLUSTER_ID_LEN];
SArray *pTableName; // table full name SArray *pTableName; // table full name
SArray *pUdf; // udf name SArray *pUdf; // udf name
bool qNodeEpset; // valid qnode bool qNodeEpset; // valid qnode
} SMetaReq; } SCatalogReq;
typedef struct SMetaData { typedef struct SCatalogRsp {
SArray *pTableMeta; // tableMeta SArray *pTableMeta; // tableMeta
SArray *pVgroupInfo; // vgroupInfo list SArray *pVgroupInfo; // vgroupInfo list
SArray *pUdfList; // udf info list SArray *pUdfList; // udf info list
SEpSet *pEpSet; // qnode epset list SEpSet *pEpSet; // qnode epset list
} SMetaData; } SCatalogRsp;
typedef struct STableComInfo { typedef struct STableComInfo {
uint8_t numOfTags; // the number of tags in schema uint8_t numOfTags; // the number of tags in schema
@ -78,29 +78,42 @@ typedef struct STableMeta {
SSchema schema[]; SSchema schema[];
} STableMeta; } STableMeta;
typedef struct SCatalogCfg {
} SCatalogCfg;
int32_t catalogInit(SCatalog *cfg);
/** /**
* Catalog service object, which is utilized to hold tableMeta (meta/vgroupInfo/udfInfo) at the client-side. * Catalog service object, which is utilized to hold tableMeta (meta/vgroupInfo/udfInfo) at the client-side.
* There is ONLY one SCatalog object for one process space, and this function returns a singleton. * There is ONLY one SCatalog object for one process space, and this function returns a singleton.
* @param pMgmtEps * @param clusterId
* @return * @return
*/ */
struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps); struct SCatalog* catalogGetHandle(const char *clusterId);
/** /**
* Get the required meta data from mnode. * Get the required meta data from mnode.
* Note that this is a synchronized API and is also thread-safety. * Note that this is a synchronized API and is also thread-safety.
* @param pCatalog * @param pCatalog
* @param pMgmtEps
* @param pMetaReq * @param pMetaReq
* @param pMetaData * @param pMetaData
* @return * @return
*/ */
int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData); int32_t catalogGetAllMeta(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, const SCatalogReq* pCatalogReq, SCatalogRsp* pCatalogData);
int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, const STableMeta* pTableMeta);
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, const STableMeta* pTableMeta, SCatalogRsp* pCatalogData);
/** /**
* Destroy catalog service handle * Destroy catalog and relase all resources
* @param pCatalog * @param pCatalog
*/ */
void destroyCatalog(struct SCatalog* pCatalog); void catalogDestroy(void);
#ifdef __cplusplus #ifdef __cplusplus
} }

8
source/libs/catalog/inc/catalogInt.h
View File

@ -23,10 +23,14 @@ extern "C" {
#include "catalog.h" #include "catalog.h"
typedef struct SCatalog { typedef struct SCatalog {
void *pMsgSender; // used to send messsage to mnode to fetch necessary metadata
SHashObj *pData; // items cached for each cluster, the hash key is the cluster-id, returned by mgmt node
} SCatalog; } SCatalog;
typedef struct SCatalogMgmt {
void *pMsgSender; // used to send messsage to mnode to fetch necessary metadata
SHashObj *pMeta; // items cached for each cluster, the hash key is the cluster-id, returned by mgmt node
} SCatalogMgmt;
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif

12
source/libs/catalog/src/catalog.c
View File

@ -15,10 +15,18 @@
#include "catalogInt.h" #include "catalogInt.h"
struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps) { SCatalogMgmt ctgMgmt = {0};
int32_t catalogInit(SCatalog *cfg) {
ctgMgmt = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
}
struct SCatalog* catalogGetHandle(const char *clusterId) {
return (struct SCatalog*) 0x1; return (struct SCatalog*) 0x1;
} }
int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) { int32_t catalogGetAllMeta(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, const SCatalogReq* pMetaReq, SCatalogRsp* pMetaData) {
return 0; return 0;
} }

4
source/libs/parser/inc/parserInt.h
View File

@ -87,13 +87,13 @@ int32_t checkForInvalidExpr(SQueryStmtInfo* pQueryInfo, SMsgBuf* pMsgBuf);
* @param msgBufLen * @param msgBufLen
* @return * @return
*/ */
int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SMetaReq* pMetaInfo, char* msg, int32_t msgBufLen); int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SCatalogReq* pMetaInfo, char* msg, int32_t msgBufLen);
/** /**
* Destroy the meta data request structure. * Destroy the meta data request structure.
* @param pMetaInfo * @param pMetaInfo
*/ */
void qParserClearupMetaRequestInfo(SMetaReq* pMetaInfo); void qParserClearupMetaRequestInfo(SCatalogReq* pMetaInfo);
#ifdef __cplusplus #ifdef __cplusplus
} }

6
source/libs/parser/src/astValidate.c
View File

@ -4077,8 +4077,8 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
} }
#endif #endif
SMetaReq req = {0}; SCatalogReq req = {0};
SMetaData data = {0}; SCatalogRsp data = {0};
// TODO: check if the qnode info has been cached already // TODO: check if the qnode info has been cached already
req.qNodeEpset = true; req.qNodeEpset = true;
@ -4088,7 +4088,7 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer
} }
// load the meta data from catalog // load the meta data from catalog
code = catalogGetMetaData(pCatalog, &req, &data); code = catalogGetAllMeta(pCatalog, NULL, &req, &data);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }

6
source/libs/parser/src/parser.c
View File

@ -42,7 +42,7 @@ int32_t qParseQuerySql(const char* pStr, size_t length, struct SQueryStmtInfo**
return TSDB_CODE_TSC_SQL_SYNTAX_ERROR; return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
} }
struct SCatalog* pCatalog = getCatalogHandle(NULL); struct SCatalog* pCatalog = catalogGetHandle(NULL);
return qParserValidateSqlNode(pCatalog, &info, *pQueryInfo, id, msg, msgLen); return qParserValidateSqlNode(pCatalog, &info, *pQueryInfo, id, msg, msgLen);
} }
@ -131,7 +131,7 @@ static void freePtrElem(void* p) {
tfree(*(char**)p); tfree(*(char**)p);
} }
int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SMetaReq* pMetaInfo, char* msg, int32_t msgBufLen) { int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SCatalogReq* pMetaInfo, char* msg, int32_t msgBufLen) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
SMsgBuf msgBuf = {.buf = msg, .len = msgBufLen}; SMsgBuf msgBuf = {.buf = msg, .len = msgBufLen};
@ -188,7 +188,7 @@ int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SMetaReq* pMet
return code; return code;
} }
void qParserClearupMetaRequestInfo(SMetaReq* pMetaReq) { void qParserClearupMetaRequestInfo(SCatalogReq* pMetaReq) {
if (pMetaReq == NULL) { if (pMetaReq == NULL) {
return; return;
} }

24
source/libs/parser/test/parserTests.cpp
View File

@ -38,7 +38,7 @@ void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_
strcpy(p->name, name); strcpy(p->name, name);
} }
void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SMetaReq* req) { void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SCatalogReq* req) {
pQueryInfo->numOfTables = 1; pQueryInfo->numOfTables = 1;
pQueryInfo->pTableMetaInfo = (STableMetaInfo**)calloc(1, POINTER_BYTES); pQueryInfo->pTableMetaInfo = (STableMetaInfo**)calloc(1, POINTER_BYTES);
@ -80,7 +80,7 @@ void sqlCheck(const char* sql, bool valid) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -117,7 +117,7 @@ TEST(testCase, validateAST_test) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -175,7 +175,7 @@ TEST(testCase, function_Test) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -221,7 +221,7 @@ TEST(testCase, function_Test2) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -267,7 +267,7 @@ TEST(testCase, function_Test3) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -312,7 +312,7 @@ TEST(testCase, function_Test4) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -360,7 +360,7 @@ TEST(testCase, function_Test5) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -445,7 +445,7 @@ TEST(testCase, function_Test6) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -523,7 +523,7 @@ TEST(testCase, function_Test6) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -585,7 +585,7 @@ TEST(testCase, function_Test6) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -664,7 +664,7 @@ TEST(testCase, function_Test6) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);

6
source/libs/parser/test/plannerTest.cpp
View File

@ -39,7 +39,7 @@ void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_
strcpy(p->name, name); strcpy(p->name, name);
} }
void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SMetaReq *req) { void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SCatalogReq *req) {
pQueryInfo->numOfTables = 1; pQueryInfo->numOfTables = 1;
pQueryInfo->pTableMetaInfo = (STableMetaInfo**)calloc(1, POINTER_BYTES); pQueryInfo->pTableMetaInfo = (STableMetaInfo**)calloc(1, POINTER_BYTES);
@ -79,7 +79,7 @@ void generateLogicplan(const char* sql) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
@ -119,7 +119,7 @@ TEST(testCase, planner_test) {
int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf);
ASSERT_EQ(code, 0); ASSERT_EQ(code, 0);
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);

2
source/libs/parser/test/tokenizerTest.cpp
View File

@ -714,7 +714,7 @@ TEST(testCase, extractMeta_test) {
ASSERT_EQ(info1.valid, true); ASSERT_EQ(info1.valid, true);
char msg[128] = {0}; char msg[128] = {0};
SMetaReq req = {0}; SCatalogReq req = {0};
int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128);
ASSERT_EQ(ret, 0); ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);