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

Merge pull request #9077 from taosdata/catalog_dev 

Catalog dev
This commit is contained in:
dapan1121 2021-12-14 15:51:18 +08:00 committed by GitHub
parent 8917dee917 3bca16804b
commit 0af02889cc
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
29 changed files with 723 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
include/common/taosmsg.h
View File

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

32
include/common/tmessage.h Normal file
View File

@ -0,0 +1,32 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#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_*/

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

@ -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_*/
#endif /*_TD_CATALOG_H_*/

8
include/util/taoserror.h
View File

@ -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

2
include/util/tlog.h
View File

@ -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

1
source/client/CMakeLists.txt
View File

@ -7,6 +7,7 @@ target_include_directories(
)
target_link_libraries(
taos
PRIVATE common
INTERFACE api
PRIVATE os util common transport parser
)

1
source/client/src/client.c
View File

@ -15,7 +15,6 @@
#include "os.h"
#include "tdef.h"
#include "tglobal.h"
#include "clientInt.h"
#include "tscLog.h"

15
source/common/inc/commonInt.h
View File

@ -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_*/
#endif /*_TD_COMMON_INT_H_*/

20
source/common/src/taosmsg.c Normal file
View File

@ -0,0 +1,20 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#define TAOS_MESSAGE_C
#include "taosmsg.h"

1
source/common/src/tglobal.c
View File

@ -1630,3 +1630,4 @@ bool taosCheckBalanceCfgOptions(const char *option, int32_t *vnodeId, int32_t *d
return true;
}

179
source/common/src/tmessage.c
View File

@ -13,6 +13,181 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#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;
*/
}

4
source/dnode/mgmt/impl/src/dndTransport.c
View File

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

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

@ -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_*/
#endif /*_TD_CATALOG_INT_H_*/

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

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

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

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

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

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

10
source/libs/parser/src/insertParser.c
View File

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

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

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

17
source/libs/parser/src/parserUtil.c
View File

@ -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));

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

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

2
source/libs/parser/test/mockCatalog.h
View File

@ -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

4
source/libs/parser/test/mockCatalogService.cpp
View File

@ -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<STableMeta> 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);
}

2
source/libs/parser/test/mockCatalogService.h
View File

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

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

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

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

@ -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);

2
source/libs/transport/src/rpcMain.c
View File

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

12
source/util/src/terror.c
View File

@ -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); }
const char* terrstr() { return tstrerror(terrno); }

2
source/util/src/tlog.c
View File

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