diff --git a/include/common/taosmsg.h b/include/common/taosmsg.h index 45f9514cf0..a75c4a7bac 100644 --- a/include/common/taosmsg.h +++ b/include/common/taosmsg.h @@ -77,7 +77,7 @@ TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_FUNCTION, "drop-function" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_CREATE_STB, "create-stb" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_ALTER_STB, "alter-stb" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_DROP_STB, "drop-stb" ) -TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_STB_VGROUP, "stb-vgroup" ) +TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_VGROUP_LIST, "vgroup-list" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_QUERY, "kill-query" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_STREAM, "kill-stream" ) TAOS_DEFINE_MESSAGE_TYPE( TSDB_MSG_TYPE_KILL_CONN, "kill-conn" ) @@ -214,6 +214,11 @@ typedef enum _mgmt_table { extern char *taosMsg[]; +typedef struct SBuildTableMetaInput { + int32_t vgId; + char *tableFullName; +} SBuildTableMetaInput; + #pragma pack(push, 1) // null-terminated string instead of char array to avoid too many memory consumption in case of more than 1M tableMeta @@ -773,9 +778,8 @@ typedef struct { } SStbInfoMsg; typedef struct { + SMsgHead msgHead; char tableFname[TSDB_TABLE_FNAME_LEN]; - int8_t createFlag; - char tags[]; } STableInfoMsg; typedef struct { @@ -790,6 +794,20 @@ typedef struct SSTableVgroupMsg { int32_t numOfTables; } SSTableVgroupMsg, SSTableVgroupRspMsg; +typedef struct SVgroupInfo { + int32_t vgId; + int8_t numOfEps; + SEpAddrMsg epAddr[TSDB_MAX_REPLICA]; +} SVgroupInfo; + +typedef struct SVgroupListRspMsg { + int32_t vgroupNum; + int32_t vgroupVersion; + SVgroupInfo vgroupInfo[]; +} SVgroupListRspMsg; + +typedef SVgroupListRspMsg SVgroupListInfo; + typedef struct { int32_t vgId; int8_t numOfEps; diff --git a/include/common/tmessage.h b/include/common/tmessage.h new file mode 100644 index 0000000000..c728ee026e --- /dev/null +++ b/include/common/tmessage.h @@ -0,0 +1,32 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + +#ifndef _TD_COMMON_TMESSAGE_H_ +#define _TD_COMMON_TMESSAGE_H_ + +#ifdef __cplusplus +extern "C" { +#endif + +extern int32_t (*tscBuildMsg[TSDB_MSG_TYPE_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen); +extern int32_t (*tscProcessMsgRsp[TSDB_MSG_TYPE_MAX])(void* output, char *msg, int32_t msgSize); + + + +#ifdef __cplusplus +} +#endif + +#endif /*_TD_COMMON_TMESSAGE_H_*/ diff --git a/include/libs/catalog/catalog.h b/include/libs/catalog/catalog.h index 050b9c904f..5019191fd1 100644 --- a/include/libs/catalog/catalog.h +++ b/include/libs/catalog/catalog.h @@ -30,16 +30,23 @@ extern "C" { struct SCatalog; -typedef struct SMetaReq { - char clusterId[TSDB_CLUSTER_ID_LEN]; +typedef struct SDBVgroupInfo { + int32_t vgroupVersion; + SArray *vgId; + int32_t hashRange; + int32_t hashNum; +} SDBVgroupInfo; + +typedef struct SCatalogReq { + char clusterId[TSDB_CLUSTER_ID_LEN]; //???? SArray *pTableName; // table full name SArray *pUdf; // udf name - bool qNodeEpset; // valid qnode -} SMetaReq; + bool qNodeRequired; // valid qnode +} SCatalogReq; typedef struct SMetaData { - SArray *pTableMeta; // tableMeta - SArray *pVgroupInfo; // vgroupInfo list + SArray *pTableMeta; // STableMeta array + SArray *pVgroupInfo; // SVgroupInfo list SArray *pUdfList; // udf info list SEpSet *pEpSet; // qnode epset list } SMetaData; @@ -78,32 +85,71 @@ typedef struct STableMeta { SSchema schema[]; } STableMeta; +typedef struct SCatalogCfg { + +} SCatalogCfg; + +int32_t catalogInit(SCatalogCfg *cfg); + /** * 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. - * @param pMgmtEps + * @param clusterId * @return */ -struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps); +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); + + + +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 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); + + +/** + * get table's vgroup list. + * @param clusterId + * @pVgroupList - array of SVgroupInfo + * @return + */ +int32_t catalogGetTableVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pTableName, SArray* pVgroupList); + /** * Get the required meta data from mnode. * Note that this is a synchronized API and is also thread-safety. * @param pCatalog + * @param pMgmtEps * @param pMetaReq * @param pMetaData * @return */ -int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData); +int32_t catalogGetAllMeta(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp); + + +int32_t catalogGetQnodeList(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, SEpSet* pQnodeEpSet); + + /** - * Destroy catalog service handle + * Destroy catalog and relase all resources * @param pCatalog */ -void destroyCatalog(struct SCatalog* pCatalog); +void catalogDestroy(void); #ifdef __cplusplus } #endif -#endif /*_TD_CATALOG_H_*/ \ No newline at end of file +#endif /*_TD_CATALOG_H_*/ diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 64a1fa84ad..8400031f9b 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -117,6 +117,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_TSC_INVALID_JSON TAOS_DEF_ERROR_CODE(0, 0x0221) //"Invalid JSON format") #define TSDB_CODE_TSC_INVALID_JSON_TYPE TAOS_DEF_ERROR_CODE(0, 0x0222) //"Invalid JSON data type") #define TSDB_CODE_TSC_VALUE_OUT_OF_RANGE TAOS_DEF_ERROR_CODE(0, 0x0223) //"Value out of range") +#define TSDB_CODE_TSC_INVALID_INPUT TAOS_DEF_ERROR_CODE(0, 0X0224) //"Invalid tsc input") // mnode #define TSDB_CODE_MND_MSG_NOT_PROCESSED TAOS_DEF_ERROR_CODE(0, 0x0300) @@ -501,6 +502,13 @@ int32_t* taosGetErrno(); // monitor #define TSDB_CODE_MON_CONNECTION_INVALID TAOS_DEF_ERROR_CODE(0, 0x2300) //"monitor invalid monitor db connection") +// catalog +#define TSDB_CODE_CTG_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2400) //catalog interval error +#define TSDB_CODE_CTG_INVALID_INPUT TAOS_DEF_ERROR_CODE(0, 0x2401) //invalid catalog input parameters +#define TSDB_CODE_CTG_NOT_READY TAOS_DEF_ERROR_CODE(0, 0x2402) //catalog is not ready +#define TSDB_CODE_CTG_MEM_ERROR TAOS_DEF_ERROR_CODE(0, 0x2403) //catalog memory error +#define TSDB_CODE_CTG_SYS_ERROR TAOS_DEF_ERROR_CODE(0, 0x2404) //catalog system error + #ifdef __cplusplus } #endif diff --git a/include/util/tlog.h b/include/util/tlog.h index 2ee60e4324..9b6033e7fe 100644 --- a/include/util/tlog.h +++ b/include/util/tlog.h @@ -45,6 +45,8 @@ extern int32_t sDebugFlag; extern int32_t tsdbDebugFlag; extern int32_t cqDebugFlag; extern int32_t debugFlag; +extern int32_t ctgDebugFlag; + #define DEBUG_FATAL 1U #define DEBUG_ERROR DEBUG_FATAL diff --git a/source/client/CMakeLists.txt b/source/client/CMakeLists.txt index 1a67faea9f..b46afa31b9 100644 --- a/source/client/CMakeLists.txt +++ b/source/client/CMakeLists.txt @@ -7,6 +7,7 @@ target_include_directories( ) target_link_libraries( taos + PRIVATE common INTERFACE api PRIVATE os util common transport parser ) diff --git a/source/client/src/client.c b/source/client/src/client.c index a99c7c44bc..863e96aa74 100644 --- a/source/client/src/client.c +++ b/source/client/src/client.c @@ -15,7 +15,6 @@ #include "os.h" #include "tdef.h" - #include "tglobal.h" #include "clientInt.h" #include "tscLog.h" diff --git a/source/common/inc/commonInt.h b/source/common/inc/commonInt.h index b8be8899f3..448600e22e 100644 --- a/source/common/inc/commonInt.h +++ b/source/common/inc/commonInt.h @@ -20,8 +20,21 @@ extern "C" { #endif + +#include "tlog.h" + +extern int32_t cDebugFlag; + +#define tscFatal(...) do { if (cDebugFlag & DEBUG_FATAL) { taosPrintLog("TSC FATAL ", cDebugFlag, __VA_ARGS__); }} while(0) +#define tscError(...) do { if (cDebugFlag & DEBUG_ERROR) { taosPrintLog("TSC ERROR ", cDebugFlag, __VA_ARGS__); }} while(0) +#define tscWarn(...) do { if (cDebugFlag & DEBUG_WARN) { taosPrintLog("TSC WARN ", cDebugFlag, __VA_ARGS__); }} while(0) +#define tscInfo(...) do { if (cDebugFlag & DEBUG_INFO) { taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); }} while(0) +#define tscDebug(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); }} while(0) +#define tscTrace(...) do { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); }} while(0) +#define tscDebugL(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC ", cDebugFlag, __VA_ARGS__); }} while(0) + #ifdef __cplusplus } #endif -#endif /*_TD_COMMON_INT_H_*/ \ No newline at end of file +#endif /*_TD_COMMON_INT_H_*/ diff --git a/source/common/src/taosmsg.c b/source/common/src/taosmsg.c new file mode 100644 index 0000000000..b35e3f1478 --- /dev/null +++ b/source/common/src/taosmsg.c @@ -0,0 +1,20 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + +#define TAOS_MESSAGE_C + +#include "taosmsg.h" + + diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 4896251a79..dfbbe33b59 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -1630,3 +1630,4 @@ bool taosCheckBalanceCfgOptions(const char *option, int32_t *vnodeId, int32_t *d return true; } + \ No newline at end of file diff --git a/source/common/src/tmessage.c b/source/common/src/tmessage.c index 0b6dbfdb51..07adf1d599 100644 --- a/source/common/src/tmessage.c +++ b/source/common/src/tmessage.c @@ -13,6 +13,181 @@ * along with this program. If not, see . */ -#define TAOS_MESSAGE_C - #include "taosmsg.h" +#include "commonint.h" + + +int32_t (*tscBuildMsg[TSDB_MSG_TYPE_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen) = {0}; + +int32_t (*tscProcessMsgRsp[TSDB_MSG_TYPE_MAX])(void* output, char *msg, int32_t msgSize) = {0}; + + +int32_t tscBuildVgroupListReqMsg(void* input, char **msg, int32_t msgSize, int32_t *msgLen) { + if (NULL == msg || NULL == msgLen) { + return TSDB_CODE_TSC_INVALID_INPUT; + } + + *msgLen = 0; + + return TSDB_CODE_SUCCESS; +} + +int32_t tscBuildTableMetaReqMsg(void* input, char **msg, int32_t msgSize, int32_t *msgLen) { + if (NULL == input || NULL == msg || NULL == msgLen) { + return TSDB_CODE_TSC_INVALID_INPUT; + } + + SBuildTableMetaInput* bInput = (SBuildTableMetaInput *)input; + + int32_t estimateSize = sizeof(STableInfoMsg); + if (NULL == *msg || msgSize < estimateSize) { + tfree(*msg); + *msg = calloc(1, estimateSize); + if (NULL == *msg) { + return TSDB_CODE_TSC_OUT_OF_MEMORY; + } + } + + STableInfoMsg *bMsg = (STableInfoMsg *)*msg; + + bMsg->msgHead.vgId = bInput->vgId; + + strncpy(bMsg->tableFname, bInput->tableFullName, sizeof(bMsg->tableFname)); + bMsg->tableFname[sizeof(bMsg->tableFname) - 1] = 0; + + *msgLen = (int32_t)sizeof(*bMsg); + + return TSDB_CODE_SUCCESS; +} + + +int32_t tscProcessVgroupListRsp(void* output, char *msg, int32_t msgSize) { + if (NULL == output || NULL == msg || msgSize <= 0) { + return TSDB_CODE_TSC_INVALID_INPUT; + } + + SVgroupListRspMsg *pRsp = (SVgroupListRspMsg *)msg; + + pRsp->vgroupNum = htonl(pRsp->vgroupNum); + pRsp->vgroupVersion = htonl(pRsp->vgroupVersion); + + if (pRsp->vgroupNum < 0) { + tscError("vgroup number[%d] in rsp is invalid", pRsp->vgroupNum); + return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE; + } + + if (pRsp->vgroupVersion < 0) { + tscError("vgroup vgroupVersion[%d] in rsp is invalid", pRsp->vgroupVersion); + return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE; + } + + if (msgSize != (pRsp->vgroupNum * sizeof(pRsp->vgroupInfo[0]) + sizeof(*pRsp))) { + tscError("vgroup list msg size mis-match, msgSize:%d, vgroup number:%d", msgSize, pRsp->vgroupNum); + return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE; + } + + // keep SVgroupListInfo/SVgroupListRspMsg the same + *(SVgroupListInfo **)output = (SVgroupListInfo *)msg; + + if (pRsp->vgroupNum == 0) { + return TSDB_CODE_SUCCESS; + } + + for (int32_t i = 0; i < pRsp->vgroupNum; ++i) { + pRsp->vgroupInfo[i].vgId = htonl(pRsp->vgroupInfo[i].vgId); + for (int32_t n = 0; n < pRsp->vgroupInfo[i].numOfEps; ++n) { + pRsp->vgroupInfo[i].epAddr[n].port = htonl(pRsp->vgroupInfo[i].epAddr[n].port); + } + } + + return TSDB_CODE_SUCCESS; +} + +void msgInit() { + tscBuildMsg[TSDB_MSG_TYPE_TABLE_META] = tscBuildTableMetaReqMsg; + tscBuildMsg[TSDB_MSG_TYPE_VGROUP_LIST] = tscBuildVgroupListReqMsg; + + + //tscProcessMsgRsp[TSDB_MSG_TYPE_TABLE_META] = tscProcessTableMetaRsp; + tscProcessMsgRsp[TSDB_MSG_TYPE_VGROUP_LIST] = tscProcessVgroupListRsp; + +/* + tscBuildMsg[TSDB_SQL_SELECT] = tscBuildQueryMsg; + tscBuildMsg[TSDB_SQL_INSERT] = tscBuildSubmitMsg; + tscBuildMsg[TSDB_SQL_FETCH] = tscBuildFetchMsg; + + tscBuildMsg[TSDB_SQL_CREATE_DB] = tscBuildCreateDbMsg; + tscBuildMsg[TSDB_SQL_CREATE_USER] = tscBuildUserMsg; + tscBuildMsg[TSDB_SQL_CREATE_FUNCTION] = tscBuildCreateFuncMsg; + + tscBuildMsg[TSDB_SQL_CREATE_ACCT] = tscBuildAcctMsg; + tscBuildMsg[TSDB_SQL_ALTER_ACCT] = tscBuildAcctMsg; + + tscBuildMsg[TSDB_SQL_CREATE_TABLE] = tscBuildCreateTableMsg; + tscBuildMsg[TSDB_SQL_DROP_USER] = tscBuildDropUserAcctMsg; + tscBuildMsg[TSDB_SQL_DROP_ACCT] = tscBuildDropUserAcctMsg; + tscBuildMsg[TSDB_SQL_DROP_DB] = tscBuildDropDbMsg; + tscBuildMsg[TSDB_SQL_DROP_FUNCTION] = tscBuildDropFuncMsg; + tscBuildMsg[TSDB_SQL_SYNC_DB_REPLICA] = tscBuildSyncDbReplicaMsg; + tscBuildMsg[TSDB_SQL_DROP_TABLE] = tscBuildDropTableMsg; + tscBuildMsg[TSDB_SQL_ALTER_USER] = tscBuildUserMsg; + tscBuildMsg[TSDB_SQL_CREATE_DNODE] = tscBuildCreateDnodeMsg; + tscBuildMsg[TSDB_SQL_DROP_DNODE] = tscBuildDropDnodeMsg; + tscBuildMsg[TSDB_SQL_CFG_DNODE] = tscBuildCfgDnodeMsg; + tscBuildMsg[TSDB_SQL_ALTER_TABLE] = tscBuildAlterTableMsg; + tscBuildMsg[TSDB_SQL_UPDATE_TAGS_VAL] = tscBuildUpdateTagMsg; + tscBuildMsg[TSDB_SQL_ALTER_DB] = tscAlterDbMsg; + tscBuildMsg[TSDB_SQL_COMPACT_VNODE] = tscBuildCompactMsg; + + tscBuildMsg[TSDB_SQL_CONNECT] = tscBuildConnectMsg; + tscBuildMsg[TSDB_SQL_USE_DB] = tscBuildUseDbMsg; + tscBuildMsg[TSDB_SQL_STABLEVGROUP] = tscBuildSTableVgroupMsg; + tscBuildMsg[TSDB_SQL_RETRIEVE_FUNC] = tscBuildRetrieveFuncMsg; + + tscBuildMsg[TSDB_SQL_HB] = tscBuildHeartBeatMsg; + tscBuildMsg[TSDB_SQL_SHOW] = tscBuildShowMsg; + tscBuildMsg[TSDB_SQL_RETRIEVE] = tscBuildRetrieveFromMgmtMsg; + tscBuildMsg[TSDB_SQL_KILL_QUERY] = tscBuildKillMsg; + tscBuildMsg[TSDB_SQL_KILL_STREAM] = tscBuildKillMsg; + tscBuildMsg[TSDB_SQL_KILL_CONNECTION] = tscBuildKillMsg; + + tscProcessMsgRsp[TSDB_SQL_SELECT] = tscProcessQueryRsp; + tscProcessMsgRsp[TSDB_SQL_FETCH] = tscProcessRetrieveRspFromNode; + + tscProcessMsgRsp[TSDB_SQL_DROP_DB] = tscProcessDropDbRsp; + tscProcessMsgRsp[TSDB_SQL_DROP_TABLE] = tscProcessDropTableRsp; + tscProcessMsgRsp[TSDB_SQL_CONNECT] = tscProcessConnectRsp; + tscProcessMsgRsp[TSDB_SQL_USE_DB] = tscProcessUseDbRsp; + tscProcessMsgRsp[TSDB_SQL_META] = tscProcessTableMetaRsp; + tscProcessMsgRsp[TSDB_SQL_STABLEVGROUP] = tscProcessSTableVgroupRsp; + tscProcessMsgRsp[TSDB_SQL_MULTI_META] = tscProcessMultiTableMetaRsp; + tscProcessMsgRsp[TSDB_SQL_RETRIEVE_FUNC] = tscProcessRetrieveFuncRsp; + + tscProcessMsgRsp[TSDB_SQL_SHOW] = tscProcessShowRsp; + tscProcessMsgRsp[TSDB_SQL_RETRIEVE] = tscProcessRetrieveRspFromNode; // rsp handled by same function. + tscProcessMsgRsp[TSDB_SQL_DESCRIBE_TABLE] = tscProcessDescribeTableRsp; + + tscProcessMsgRsp[TSDB_SQL_CURRENT_DB] = tscProcessLocalRetrieveRsp; + tscProcessMsgRsp[TSDB_SQL_CURRENT_USER] = tscProcessLocalRetrieveRsp; + tscProcessMsgRsp[TSDB_SQL_SERV_VERSION] = tscProcessLocalRetrieveRsp; + tscProcessMsgRsp[TSDB_SQL_CLI_VERSION] = tscProcessLocalRetrieveRsp; + tscProcessMsgRsp[TSDB_SQL_SERV_STATUS] = tscProcessLocalRetrieveRsp; + + tscProcessMsgRsp[TSDB_SQL_RETRIEVE_EMPTY_RESULT] = tscProcessEmptyResultRsp; + + tscProcessMsgRsp[TSDB_SQL_RETRIEVE_GLOBALMERGE] = tscProcessRetrieveGlobalMergeRsp; + + tscProcessMsgRsp[TSDB_SQL_ALTER_TABLE] = tscProcessAlterTableMsgRsp; + tscProcessMsgRsp[TSDB_SQL_ALTER_DB] = tscProcessAlterDbMsgRsp; + tscProcessMsgRsp[TSDB_SQL_COMPACT_VNODE] = tscProcessCompactRsp; + + tscProcessMsgRsp[TSDB_SQL_SHOW_CREATE_TABLE] = tscProcessShowCreateRsp; + tscProcessMsgRsp[TSDB_SQL_SHOW_CREATE_STABLE] = tscProcessShowCreateRsp; + tscProcessMsgRsp[TSDB_SQL_SHOW_CREATE_DATABASE] = tscProcessShowCreateRsp; +*/ +} + + + + + diff --git a/source/dnode/mgmt/impl/src/dndTransport.c b/source/dnode/mgmt/impl/src/dndTransport.c index 0c4eca569d..6dc46cefcd 100644 --- a/source/dnode/mgmt/impl/src/dndTransport.c +++ b/source/dnode/mgmt/impl/src/dndTransport.c @@ -72,7 +72,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) { pMgmt->msgFp[TSDB_MSG_TYPE_CREATE_STB] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_ALTER_STB] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_DROP_STB] = dndProcessMnodeWriteMsg; - pMgmt->msgFp[TSDB_MSG_TYPE_STB_VGROUP] = dndProcessMnodeReadMsg; + pMgmt->msgFp[TSDB_MSG_TYPE_VGROUP_LIST] = dndProcessMnodeReadMsg; pMgmt->msgFp[TSDB_MSG_TYPE_KILL_QUERY] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_KILL_STREAM] = dndProcessMnodeWriteMsg; pMgmt->msgFp[TSDB_MSG_TYPE_KILL_CONN] = dndProcessMnodeWriteMsg; @@ -368,4 +368,4 @@ void dndSendMsgToMnode(SDnode *pDnode, SRpcMsg *pMsg) { SEpSet epSet = {0}; dndGetMnodeEpSet(pDnode, &epSet); dndSendMsgToDnode(pDnode, &epSet, pMsg); -} \ No newline at end of file +} diff --git a/source/libs/catalog/inc/catalogInt.h b/source/libs/catalog/inc/catalogInt.h index 5b50bbff4c..720f197782 100644 --- a/source/libs/catalog/inc/catalogInt.h +++ b/source/libs/catalog/inc/catalogInt.h @@ -21,14 +21,58 @@ extern "C" { #endif #include "catalog.h" +#include "common.h" +#include "tlog.h" + +#define CTG_DEFAULT_CLUSTER_NUMBER 6 +#define CTG_DEFAULT_VGROUP_NUMBER 100 + +#define CTG_DEFAULT_INVALID_VERSION (-1) + +typedef struct SVgroupListCache { + int32_t vgroupVersion; + SHashObj *cache; // key:vgId, value:SVgroupInfo* + SArray *arrayCache; // SVgroupInfo +} SVgroupListCache; + +typedef struct SDBVgroupCache { + SHashObj *cache; //key:dbname, value:SDBVgroupInfo +} SDBVgroupCache; + +typedef struct STableMetaCache { + SHashObj *cache; //key:fulltablename, value:STableMeta +} STableMetaCache; 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 + SVgroupListCache vgroupCache; + 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 +} SCatalogMgmt; + + +extern int32_t ctgDebugFlag; + +#define ctgFatal(...) do { if (ctgDebugFlag & DEBUG_FATAL) { taosPrintLog("CTG FATAL ", ctgDebugFlag, __VA_ARGS__); }} while(0) +#define ctgError(...) do { if (ctgDebugFlag & DEBUG_ERROR) { taosPrintLog("CTG ERROR ", ctgDebugFlag, __VA_ARGS__); }} while(0) +#define ctgWarn(...) do { if (ctgDebugFlag & DEBUG_WARN) { taosPrintLog("CTG WARN ", ctgDebugFlag, __VA_ARGS__); }} while(0) +#define ctgInfo(...) do { if (ctgDebugFlag & DEBUG_INFO) { taosPrintLog("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0) +#define ctgDebug(...) do { if (ctgDebugFlag & DEBUG_DEBUG) { taosPrintLog("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0) +#define ctgTrace(...) do { if (ctgDebugFlag & DEBUG_TRACE) { taosPrintLog("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0) +#define ctgDebugL(...) do { if (ctgDebugFlag & DEBUG_DEBUG) { taosPrintLongString("CTG ", ctgDebugFlag, __VA_ARGS__); }} while(0) + + +#define CTG_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { return _code; } } while (0) +#define CTG_ERR_LRET(c,...) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { ctgError(__VA_ARGS__); return _code; } } while (0) +#define CTG_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { goto _return; } } while (0) + + #ifdef __cplusplus } #endif -#endif /*_TD_CATALOG_INT_H_*/ \ No newline at end of file +#endif /*_TD_CATALOG_INT_H_*/ diff --git a/source/libs/catalog/src/catalog.c b/source/libs/catalog/src/catalog.c index 08e2172adb..f4a4b01dab 100644 --- a/source/libs/catalog/src/catalog.c +++ b/source/libs/catalog/src/catalog.c @@ -14,11 +14,294 @@ */ #include "catalogInt.h" +#include "trpc.h" +#include "tmessage.h" -struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps) { - return (struct SCatalog*) 0x1; +SCatalogMgmt ctgMgmt = {0}; + +int32_t ctgGetVgroupFromMnode(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SVgroupListInfo** pVgroup) { + char *msg = NULL; + SEpSet *pVnodeEpSet = NULL; + int32_t msgLen = 0; + + int32_t code = tscBuildMsg[TSDB_MSG_TYPE_VGROUP_LIST](NULL, &msg, 0, &msgLen); + if (code) { + return code; + } + + SRpcMsg rpcMsg = { + .msgType = TSDB_MSG_TYPE_VGROUP_LIST, + .pCont = msg, + .contLen = msgLen, + }; + + SRpcMsg rpcRsp = {0}; + + rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp); + + code = tscProcessMsgRsp[TSDB_MSG_TYPE_VGROUP_LIST](pVgroup, rpcRsp.pCont, rpcRsp.contLen); + if (code) { + return code; + } + + return TSDB_CODE_SUCCESS; } -int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) { +int32_t ctgGetVgroupFromCache(SCatalog* pCatalog, SArray** pVgroupList, int32_t* exist) { + if (NULL == pCatalog->vgroupCache.arrayCache || pCatalog->vgroupCache.vgroupVersion < 0) { + *exist = 0; + return TSDB_CODE_SUCCESS; + } + + if (pVgroupList) { + *pVgroupList = taosArrayDup(pCatalog->vgroupCache.arrayCache); + } + + *exist = 1; + + return TSDB_CODE_SUCCESS; +} + + + + + +int32_t catalogInit(SCatalogCfg *cfg) { + ctgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK); + if (NULL == ctgMgmt.pCluster) { + CTG_ERR_LRET(TSDB_CODE_CTG_INTERNAL_ERROR, "init %d cluster cache failed", CTG_DEFAULT_CLUSTER_NUMBER); + } + + return TSDB_CODE_SUCCESS; +} + + +int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle) { + if (NULL == clusterId || NULL == catalogHandle) { + return TSDB_CODE_CTG_INVALID_INPUT; + } + + if (NULL == ctgMgmt.pCluster) { + ctgError("cluster cache are not ready"); + return TSDB_CODE_CTG_NOT_READY; + } + + size_t clen = strlen(clusterId); + SCatalog *clusterCtg = (SCatalog *)taosHashGet(ctgMgmt.pCluster, clusterId, clen); + + if (clusterCtg) { + *catalogHandle = clusterCtg; + return TSDB_CODE_SUCCESS; + } + + clusterCtg = calloc(1, sizeof(*clusterCtg)); + if (NULL == clusterCtg) { + ctgError("calloc %d failed", (int32_t)sizeof(*clusterCtg)); + return TSDB_CODE_CTG_MEM_ERROR; + } + + clusterCtg->vgroupCache.vgroupVersion = CTG_DEFAULT_INVALID_VERSION; + + if (taosHashPut(ctgMgmt.pCluster, clusterId, clen, &clusterCtg, POINTER_BYTES)) { + ctgError("put cluster %s cache to hash failed", clusterId); + tfree(clusterCtg); + return TSDB_CODE_CTG_INTERNAL_ERROR; + } + + *catalogHandle = clusterCtg; + + return TSDB_CODE_SUCCESS; +} + + +int32_t catalogGetVgroupVersion(struct SCatalog* pCatalog, int32_t* version) { + if (NULL == pCatalog || NULL == version) { + return TSDB_CODE_CTG_INVALID_INPUT; + } + + *version = pCatalog->vgroupCache.vgroupVersion; + + return TSDB_CODE_SUCCESS; +} + + + +int32_t catalogUpdateVgroup(struct SCatalog* pCatalog, SVgroupListInfo* pVgroup) { + if (NULL == pVgroup) { + ctgError("vgroup get from mnode succeed, but no output"); + return TSDB_CODE_CTG_INTERNAL_ERROR; + } + + if (pVgroup->vgroupVersion < 0) { + ctgError("vgroup version[%d] is invalid", pVgroup->vgroupVersion); + return TSDB_CODE_CTG_INVALID_INPUT; + } + + + if (NULL == pCatalog->vgroupCache.arrayCache) { + pCatalog->vgroupCache.arrayCache = taosArrayInit(pVgroup->vgroupNum, sizeof(pVgroup->vgroupInfo[0])); + if (NULL == pCatalog->vgroupCache.arrayCache) { + ctgError("init array[%d] for cluster cache failed", pVgroup->vgroupNum); + return TSDB_CODE_CTG_MEM_ERROR; + } + } else { + taosArrayClear(pCatalog->vgroupCache.arrayCache); + } + + if (NULL == pCatalog->vgroupCache.cache) { + pCatalog->vgroupCache.cache = taosHashInit(CTG_DEFAULT_VGROUP_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK); + if (NULL == pCatalog->vgroupCache.cache) { + ctgError("init hash[%d] for cluster cache failed", CTG_DEFAULT_VGROUP_NUMBER); + return TSDB_CODE_CTG_MEM_ERROR; + } + } else { + taosHashClear(pCatalog->vgroupCache.cache); + } + + SVgroupInfo *vInfo = NULL; + for (int32_t i = 0; i < pVgroup->vgroupNum; ++i) { + vInfo = taosArrayPush(pCatalog->vgroupCache.arrayCache, &pVgroup->vgroupInfo[i]); + if (NULL == vInfo) { + ctgError("push to vgroup array cache failed"); + goto error_exit; + } + + if (taosHashPut(pCatalog->vgroupCache.cache, &pVgroup->vgroupInfo[i].vgId, sizeof(pVgroup->vgroupInfo[i].vgId), &vInfo, POINTER_BYTES) != 0) { + ctgError("push to vgroup hash cache failed"); + goto error_exit; + } + } + + pCatalog->vgroupCache.vgroupVersion = pVgroup->vgroupVersion; + + return TSDB_CODE_SUCCESS; + +error_exit: + if (pCatalog->vgroupCache.arrayCache) { + taosArrayDestroy(pCatalog->vgroupCache.arrayCache); + pCatalog->vgroupCache.arrayCache = NULL; + } + + if (pCatalog->vgroupCache.cache) { + taosHashCleanup(pCatalog->vgroupCache.cache); + pCatalog->vgroupCache.cache = NULL; + } + + pCatalog->vgroupCache.vgroupVersion = CTG_DEFAULT_INVALID_VERSION; + + return TSDB_CODE_CTG_INTERNAL_ERROR; +} + + +int32_t catalogGetVgroup(SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray** pVgroupList) { + if (NULL == pCatalog || NULL == pMgmtEps || NULL == pRpc) { + return TSDB_CODE_CTG_INVALID_INPUT; + } + + int32_t exist = 0; + + CTG_ERR_RET(ctgGetVgroupFromCache(pCatalog, pVgroupList, &exist)); + + if (exist) { + return TSDB_CODE_SUCCESS; + } + + SVgroupListInfo *pVgroup = NULL; + + CTG_ERR_RET(ctgGetVgroupFromMnode(pCatalog, pRpc, pMgmtEps, &pVgroup)); + + CTG_ERR_RET(catalogUpdateVgroup(pCatalog, pVgroup)); + + if (pVgroupList) { + CTG_ERR_RET(ctgGetVgroupFromCache(pCatalog, pVgroupList, &exist)); + } + + if (0 == exist) { + ctgError("catalog fetched but get from cache failed"); + return TSDB_CODE_CTG_INTERNAL_ERROR; + } + + return TSDB_CODE_SUCCESS; +} + +int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version) { + if (NULL == pCatalog || NULL == dbName || NULL == version) { + return TSDB_CODE_CTG_INVALID_INPUT; + } + + if (NULL == pCatalog->dbCache.cache) { + *version = CTG_DEFAULT_INVALID_VERSION; + return TSDB_CODE_SUCCESS; + } + + SDBVgroupInfo * dbInfo = taosHashGet(pCatalog->dbCache.cache, dbName, strlen(dbName)); + if (NULL == dbInfo) { + *version = CTG_DEFAULT_INVALID_VERSION; + return TSDB_CODE_SUCCESS; + } + + *version = dbInfo->vgroupVersion; + + return TSDB_CODE_SUCCESS; +} + +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 catalogGetTableMetaFromMnode(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pTableName, const STagData* tagData, STableMeta* pTableMeta) { + if (NULL == pCatalog || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) { + return TSDB_CODE_CTG_INVALID_INPUT; + } + + SBuildTableMetaInput bInput = {0}; + char *msg = NULL; + SEpSet *pVnodeEpSet = NULL; + int32_t msgLen = 0; + + int32_t code = tscBuildMsg[TSDB_MSG_TYPE_TABLE_META](&bInput, &msg, 0, &msgLen); + if (code) { + return code; + } + + SRpcMsg rpcMsg = { + .msgType = TSDB_MSG_TYPE_TABLE_META, + .pCont = msg, + .contLen = msgLen, + }; + + SRpcMsg rpcRsp = {0}; + + rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp); + + return TSDB_CODE_SUCCESS; +} + +int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* pTableName, STableMeta* pTableMeta) { + +} + + +int32_t catalogGetAllMeta(struct SCatalog* pCatalog, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp) { + if (NULL == pCatalog || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) { + return TSDB_CODE_CTG_INVALID_INPUT; + } + return 0; } + +void catalogDestroy(void) { + if (ctgMgmt.pCluster) { + taosHashCleanup(ctgMgmt.pCluster); //TBD + ctgMgmt.pCluster = NULL; + } +} + + + diff --git a/source/libs/parser/inc/parserInt.h b/source/libs/parser/inc/parserInt.h index 47f655d446..e3925d3446 100644 --- a/source/libs/parser/inc/parserInt.h +++ b/source/libs/parser/inc/parserInt.h @@ -79,13 +79,13 @@ int32_t checkForInvalidExpr(SQueryStmtInfo* pQueryInfo, SMsgBuf* pMsgBuf); * @param msgBufLen * @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. * @param pMetaInfo */ -void qParserClearupMetaRequestInfo(SMetaReq* pMetaInfo); +void qParserClearupMetaRequestInfo(SCatalogReq* pMetaInfo); #ifdef __cplusplus } diff --git a/source/libs/parser/src/astValidate.c b/source/libs/parser/src/astValidate.c index 7b6c423fb6..d2813886b3 100644 --- a/source/libs/parser/src/astValidate.c +++ b/source/libs/parser/src/astValidate.c @@ -4077,18 +4077,18 @@ int32_t qParserValidateSqlNode(struct SCatalog* pCatalog, SSqlInfo* pInfo, SQuer } #endif - SMetaReq req = {0}; + SCatalogReq req = {0}; SMetaData data = {0}; // TODO: check if the qnode info has been cached already - req.qNodeEpset = true; + req.qNodeRequired = true; code = qParserExtractRequestedMetaInfo(pInfo, &req, msgBuf, msgBufLen); if (code != TSDB_CODE_SUCCESS) { return code; } // load the meta data from catalog - code = catalogGetMetaData(pCatalog, &req, &data); + code = catalogGetAllMeta(pCatalog, NULL, &req, &data); if (code != TSDB_CODE_SUCCESS) { return code; } diff --git a/source/libs/parser/src/insertParser.c b/source/libs/parser/src/insertParser.c index c6ae149f1f..fa59bc6ca7 100644 --- a/source/libs/parser/src/insertParser.c +++ b/source/libs/parser/src/insertParser.c @@ -178,15 +178,15 @@ static int32_t buildTableName(SInsertParseContext* pCxt, SToken* pStname, SArray return TSDB_CODE_SUCCESS; } -static int32_t buildMetaReq(SInsertParseContext* pCxt, SToken* pStname, SMetaReq* pMetaReq) { +static int32_t buildMetaReq(SInsertParseContext* pCxt, SToken* pStname, SCatalogReq* pMetaReq) { pMetaReq->pTableName = taosArrayInit(4, sizeof(SName)); return buildTableName(pCxt, pStname, pMetaReq->pTableName); } static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) { - SMetaReq req; + SCatalogReq req; CHECK_CODE(buildMetaReq(pCxt, pTname, &req)); - CHECK_CODE(catalogGetMetaData(pCxt->pCatalog, &req, &pCxt->meta)); + CHECK_CODE(catalogGetTableMeta(pCxt->pCatalog, NULL, NULL, NULL, &pCxt->meta)); //TODO pCxt->pTableMeta = (STableMeta*)taosArrayGetP(pCxt->meta.pTableMeta, 0); return TSDB_CODE_SUCCESS; } @@ -861,13 +861,15 @@ int32_t parseInsertSql(SParseContext* pContext, SInsertStmtInfo** pInfo) { .pComCxt = pContext, .pSql = pContext->pSql, .msg = {.buf = pContext->pMsg, .len = pContext->msgLen}, - .pCatalog = getCatalogHandle(pContext->pEpSet), + .pCatalog = NULL, .pTableMeta = NULL, .pTableBlockHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false), .totalNum = 0, .pOutput = *pInfo }; + CHECK_CODE(catalogGetHandle(NULL, &context.pCatalog)); //TODO + if (NULL == context.pTableBlockHashObj) { return TSDB_CODE_TSC_OUT_OF_MEMORY; } diff --git a/source/libs/parser/src/parser.c b/source/libs/parser/src/parser.c index a8642e2535..d167f7ad3c 100644 --- a/source/libs/parser/src/parser.c +++ b/source/libs/parser/src/parser.c @@ -43,7 +43,12 @@ int32_t qParseQuerySql(const char* pStr, size_t length, struct SQueryStmtInfo** return TSDB_CODE_TSC_SQL_SYNTAX_ERROR; } - struct SCatalog* pCatalog = getCatalogHandle(NULL); + struct SCatalog* pCatalog = NULL; + int32_t code = catalogGetHandle(NULL, &pCatalog); + if (code) { + return code; + } + return qParserValidateSqlNode(pCatalog, &info, *pQueryInfo, id, msg, msgLen); } @@ -132,7 +137,7 @@ static void freePtrElem(void* 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; SMsgBuf msgBuf = {.buf = msg, .len = msgBufLen}; @@ -189,7 +194,7 @@ int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SMetaReq* pMet return code; } -void qParserClearupMetaRequestInfo(SMetaReq* pMetaReq) { +void qParserClearupMetaRequestInfo(SCatalogReq* pMetaReq) { if (pMetaReq == NULL) { return; } diff --git a/source/libs/parser/src/parserUtil.c b/source/libs/parser/src/parserUtil.c index 7b3239c0dd..28d01b9e66 100644 --- a/source/libs/parser/src/parserUtil.c +++ b/source/libs/parser/src/parserUtil.c @@ -1448,23 +1448,6 @@ void* vgroupInfoClear(SVgroupsInfo *vgroupList) { return NULL; } -char* serializeTagData(STagData* pTagData, char* pMsg) { - int32_t n = (int32_t) strlen(pTagData->name); - *(int32_t*) pMsg = htonl(n); - pMsg += sizeof(n); - - memcpy(pMsg, pTagData->name, n); - pMsg += n; - - *(int32_t*)pMsg = htonl(pTagData->dataLen); - pMsg += sizeof(int32_t); - - memcpy(pMsg, pTagData->data, pTagData->dataLen); - pMsg += pTagData->dataLen; - - return pMsg; -} - int32_t copyTagData(STagData* dst, const STagData* src) { dst->dataLen = src->dataLen; tstrncpy(dst->name, src->name, tListLen(dst->name)); diff --git a/source/libs/parser/test/mockCatalog.cpp b/source/libs/parser/test/mockCatalog.cpp index 1f6fd71030..f2b34971ef 100644 --- a/source/libs/parser/test/mockCatalog.cpp +++ b/source/libs/parser/test/mockCatalog.cpp @@ -48,6 +48,6 @@ struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps) { return mockCatalogService->getCatalogHandle(pMgmtEps); } -int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) { +int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SCatalogReq* pMetaReq, SMetaData* pMetaData) { return mockCatalogService->catalogGetMetaData(pCatalog, pMetaReq, pMetaData); } diff --git a/source/libs/parser/test/mockCatalog.h b/source/libs/parser/test/mockCatalog.h index 32001d19fd..e60f7727e6 100644 --- a/source/libs/parser/test/mockCatalog.h +++ b/source/libs/parser/test/mockCatalog.h @@ -22,6 +22,6 @@ void generateMetaData(MockCatalogService* mcs); // mock struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps); -int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData); +int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SCatalogReq* pMetaReq, SMetaData* pMetaData); #endif // MOCK_CATALOG_H diff --git a/source/libs/parser/test/mockCatalogService.cpp b/source/libs/parser/test/mockCatalogService.cpp index 46e82566d6..457f8d88bd 100644 --- a/source/libs/parser/test/mockCatalogService.cpp +++ b/source/libs/parser/test/mockCatalogService.cpp @@ -87,7 +87,7 @@ public: return (struct SCatalog*)0x01; } - int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) const { + int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SCatalogReq* pMetaReq, SMetaData* pMetaData) const { assert(nullptr != pMetaReq && 1 == taosArrayGetSize(pMetaReq->pTableName)); SName* fullName = (SName*)taosArrayGet(pMetaReq->pTableName, 0); std::unique_ptr table; @@ -248,7 +248,7 @@ struct SCatalog* MockCatalogService::getCatalogHandle(const SEpSet* pMgmtEps) co return impl_->getCatalogHandle(pMgmtEps); } -int32_t MockCatalogService::catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) const { +int32_t MockCatalogService::catalogGetMetaData(struct SCatalog* pCatalog, const SCatalogReq* pMetaReq, SMetaData* pMetaData) const { return impl_->catalogGetMetaData(pCatalog, pMetaReq, pMetaData); } diff --git a/source/libs/parser/test/mockCatalogService.h b/source/libs/parser/test/mockCatalogService.h index 79572086a1..66b439b3e9 100644 --- a/source/libs/parser/test/mockCatalogService.h +++ b/source/libs/parser/test/mockCatalogService.h @@ -50,7 +50,7 @@ public: MockCatalogService(); ~MockCatalogService(); struct SCatalog* getCatalogHandle(const SEpSet* pMgmtEps) const; - int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SMetaReq* pMetaReq, SMetaData* pMetaData) const; + int32_t catalogGetMetaData(struct SCatalog* pCatalog, const SCatalogReq* pMetaReq, SMetaData* pMetaData) const; ITableBuilder& createTableBuilder(const std::string& db, const std::string& tbname, int8_t tableType, int32_t numOfColumns, int32_t numOfTags = 0); void createSubTable(const std::string& db, const std::string& stbname, const std::string& tbname, int16_t vgid); void showTables() const; diff --git a/source/libs/parser/test/parserTests.cpp b/source/libs/parser/test/parserTests.cpp index 2193a44604..6a402cd620 100644 --- a/source/libs/parser/test/parserTests.cpp +++ b/source/libs/parser/test/parserTests.cpp @@ -38,7 +38,7 @@ void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_ strcpy(p->name, name); } -void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SMetaReq* req) { +void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SCatalogReq* req) { pQueryInfo->numOfTables = 1; 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); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -117,7 +117,7 @@ TEST(testCase, validateAST_test) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -175,7 +175,7 @@ TEST(testCase, function_Test) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -221,7 +221,7 @@ TEST(testCase, function_Test2) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -267,7 +267,7 @@ TEST(testCase, function_Test3) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -312,7 +312,7 @@ TEST(testCase, function_Test4) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -360,7 +360,7 @@ TEST(testCase, function_Test5) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -445,7 +445,7 @@ TEST(testCase, function_Test6) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -523,7 +523,7 @@ TEST(testCase, function_Test6) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -585,7 +585,7 @@ TEST(testCase, function_Test6) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -664,7 +664,7 @@ TEST(testCase, function_Test6) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); diff --git a/source/libs/parser/test/plannerTest.cpp b/source/libs/parser/test/plannerTest.cpp index c86e687664..bb9271a3c8 100644 --- a/source/libs/parser/test/plannerTest.cpp +++ b/source/libs/parser/test/plannerTest.cpp @@ -39,7 +39,7 @@ void setSchema(SSchema* p, int32_t type, int32_t bytes, const char* name, int32_ strcpy(p->name, name); } -void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SMetaReq *req) { +void setTableMetaInfo(SQueryStmtInfo* pQueryInfo, SCatalogReq *req) { pQueryInfo->numOfTables = 1; 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); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); @@ -119,7 +119,7 @@ TEST(testCase, planner_test) { int32_t code = evaluateSqlNode(pNode, TSDB_TIME_PRECISION_NANO, &buf); ASSERT_EQ(code, 0); - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); diff --git a/source/libs/parser/test/tokenizerTest.cpp b/source/libs/parser/test/tokenizerTest.cpp index 032a93cd32..23c0aae15f 100644 --- a/source/libs/parser/test/tokenizerTest.cpp +++ b/source/libs/parser/test/tokenizerTest.cpp @@ -709,7 +709,7 @@ TEST(testCase, extractMeta_test) { ASSERT_EQ(info1.valid, true); char msg[128] = {0}; - SMetaReq req = {0}; + SCatalogReq req = {0}; int32_t ret = qParserExtractRequestedMetaInfo(&info1, &req, msg, 128); ASSERT_EQ(ret, 0); ASSERT_EQ(taosArrayGetSize(req.pTableName), 1); diff --git a/source/libs/transport/src/rpcMain.c b/source/libs/transport/src/rpcMain.c index c54415860a..2f710f3efd 100644 --- a/source/libs/transport/src/rpcMain.c +++ b/source/libs/transport/src/rpcMain.c @@ -411,7 +411,7 @@ void rpcSendRequest(void *shandle, const SEpSet *pEpSet, SRpcMsg *pMsg, int64_t // for TDengine, all the query, show commands shall have TCP connection char type = pMsg->msgType; if (type == TSDB_MSG_TYPE_QUERY || type == TSDB_MSG_TYPE_SHOW_RETRIEVE - || type == TSDB_MSG_TYPE_FETCH || type == TSDB_MSG_TYPE_STB_VGROUP + || type == TSDB_MSG_TYPE_FETCH || type == TSDB_MSG_TYPE_VGROUP_LIST || type == TSDB_MSG_TYPE_TABLES_META || type == TSDB_MSG_TYPE_TABLE_META || type == TSDB_MSG_TYPE_SHOW || type == TSDB_MSG_TYPE_STATUS || type == TSDB_MSG_TYPE_ALTER_TABLE) pContext->connType = RPC_CONN_TCPC; diff --git a/source/util/src/terror.c b/source/util/src/terror.c index ceb344ffe8..42fde042e7 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -127,6 +127,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSC_DUP_TAG_NAMES, "duplicated tag names" TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_JSON, "Invalid JSON format") TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_JSON_TYPE, "Invalid JSON data type") TAOS_DEFINE_ERROR(TSDB_CODE_TSC_VALUE_OUT_OF_RANGE, "Value out of range") +TAOS_DEFINE_ERROR(TSDB_CODE_TSC_INVALID_INPUT, "Invalid tsc input") // mnode TAOS_DEFINE_ERROR(TSDB_CODE_MND_MSG_NOT_PROCESSED, "Message not processed") @@ -497,6 +498,15 @@ TAOS_DEFINE_ERROR(TSDB_CODE_FS_FILE_ALREADY_EXISTS, "tfs file already exis TAOS_DEFINE_ERROR(TSDB_CODE_FS_INVLD_LEVEL, "tfs invalid level") TAOS_DEFINE_ERROR(TSDB_CODE_FS_NO_VALID_DISK, "tfs no valid disk") +// catalog +TAOS_DEFINE_ERROR(TSDB_CODE_CTG_INTERNAL_ERROR, "catalog interval error") +TAOS_DEFINE_ERROR(TSDB_CODE_CTG_INVALID_INPUT, "invalid catalog input parameters") +TAOS_DEFINE_ERROR(TSDB_CODE_CTG_NOT_READY, "catalog is not ready") +TAOS_DEFINE_ERROR(TSDB_CODE_CTG_MEM_ERROR, "catalog memory error") +TAOS_DEFINE_ERROR(TSDB_CODE_CTG_SYS_ERROR, "catalog system error") + + + #ifdef TAOS_ERROR_C }; #endif @@ -545,4 +555,4 @@ const char* tstrerror(int32_t err) { return ""; } -const char* terrstr() { return tstrerror(terrno); } \ No newline at end of file +const char* terrstr() { return tstrerror(terrno); } diff --git a/source/util/src/tlog.c b/source/util/src/tlog.c index 24293fe9d3..2d7179c0d3 100644 --- a/source/util/src/tlog.c +++ b/source/util/src/tlog.c @@ -99,6 +99,8 @@ int32_t wDebugFlag = 135; int32_t tsdbDebugFlag = 131; int32_t cqDebugFlag = 131; int32_t fsDebugFlag = 135; +int32_t ctgDebugFlag = 131; + int64_t dbgEmptyW = 0; int64_t dbgWN = 0;