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

Merge remote-tracking branch 'origin/3.0' into feature/dnode3

This commit is contained in:
Shengliang Guan 2022-01-16 19:40:24 -08:00
parent de8b6e7351 1b1a6178d5
commit ae2d77e922
195 changed files with 4566 additions and 4018 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.devcontainer/devcontainer.json
View File

@ -26,7 +26,8 @@
"eamodio.gitlens",
"matepek.vscode-catch2-test-adapter",
"spmeesseman.vscode-taskexplorer",
"cschlosser.doxdocgen"
"cschlosser.doxdocgen",
"urosvujosevic.explorer-manager"
],
// Use 'forwardPorts' to make a list of ports inside the container available locally.
// "forwardPorts": [],

8
cmake/libuv_CMakeLists.txt.in
View File

@ -4,9 +4,9 @@ ExternalProject_Add(libuv
GIT_REPOSITORY https://github.com/libuv/libuv.git
GIT_TAG v1.42.0
SOURCE_DIR "${CMAKE_CONTRIB_DIR}/libuv"
BINARY_DIR ""
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
BINARY_DIR "${CMAKE_CONTRIB_DIR}/libuv"
CONFIGURE_COMMAND ""
BUILD_COMMAND ""
INSTALL_COMMAND ""
TEST_COMMAND ""
)
)

3
include/client/taos.h
View File

@ -193,8 +193,7 @@ DLL_EXPORT void taos_close_stream(TAOS_STREAM *tstr);
DLL_EXPORT int taos_load_table_info(TAOS *taos, const char* tableNameList);
DLL_EXPORT TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision);
DLL_EXPORT TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sqlLen);
DLL_EXPORT TAOS_RES *taos_create_topic(TAOS* taos, const char* name, const char* sql, int sqlLen);
#ifdef __cplusplus
}

60
include/common/tmsg.h
View File

@ -68,6 +68,14 @@ typedef uint16_t tmsg_t;
#define TSDB_IE_TYPE_DNODE_EXT 6
#define TSDB_IE_TYPE_DNODE_STATE 7
typedef enum {
HEARTBEAT_TYPE_MQ = 0,
HEARTBEAT_TYPE_QUERY = 1,
// types can be added here
//
HEARTBEAT_TYPE_MAX
} EHbType;
typedef enum _mgmt_table {
TSDB_MGMT_TABLE_START,
TSDB_MGMT_TABLE_ACCT,
@ -147,7 +155,7 @@ typedef struct {
typedef struct {
SClientHbKey connKey;
SHashObj* info; // hash<Slv.key, Sklv>
SHashObj* info; // hash<Skv.key, Skv>
} SClientHbReq;
typedef struct {
@ -173,7 +181,10 @@ static FORCE_INLINE uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) {
}
int tSerializeSClientHbReq(void** buf, const SClientHbReq* pReq);
void* tDeserializeClientHbReq(void* buf, SClientHbReq* pReq);
void* tDeserializeSClientHbReq(void* buf, SClientHbReq* pReq);
int tSerializeSClientHbRsp(void** buf, const SClientHbRsp* pRsp);
void* tDeserializeSClientHbRsp(void* buf, SClientHbRsp* pRsp);
static FORCE_INLINE void tFreeClientHbReq(void *pReq) {
SClientHbReq* req = (SClientHbReq*)pReq;
@ -182,14 +193,17 @@ static FORCE_INLINE void tFreeClientHbReq(void *pReq) {
}
int tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq);
void* tDeserializeClientHbBatchReq(void* buf, SClientHbBatchReq* pReq);
void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pReq);
static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq) {
SClientHbBatchReq *req = (SClientHbBatchReq*)pReq;
taosArrayDestroyEx(req->reqs, tFreeClientHbReq);
//taosArrayDestroyEx(req->reqs, tFreeClientHbReq);
free(pReq);
}
int tSerializeSClientHbBatchRsp(void** buf, const SClientHbBatchRsp* pBatchRsp);
void* tDeserializeSClientHbBatchRsp(void* buf, SClientHbBatchRsp* pBatchRsp);
static FORCE_INLINE int taosEncodeSKv(void** buf, const SKv* pKv) {
int tlen = 0;
tlen += taosEncodeFixedI32(buf, pKv->keyLen);
@ -220,6 +234,7 @@ static FORCE_INLINE void* taosDecodeSClientHbKey(void* buf, SClientHbKey* pKey)
return buf;
}
typedef struct {
int32_t vgId;
char* dbName;
@ -359,6 +374,31 @@ static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEp) {
return buf;
}
typedef struct SMqHbRsp {
int8_t status; //idle or not
int8_t vnodeChanged;
int8_t epChanged; // should use new epset
int8_t reserved;
SEpSet epSet;
} SMqHbRsp;
static FORCE_INLINE int taosEncodeSMqHbRsp(void** buf, const SMqHbRsp* pRsp) {
int tlen = 0;
tlen += taosEncodeFixedI8(buf, pRsp->status);
tlen += taosEncodeFixedI8(buf, pRsp->vnodeChanged);
tlen += taosEncodeFixedI8(buf, pRsp->epChanged);
tlen += taosEncodeSEpSet(buf, &pRsp->epSet);
return tlen;
}
static FORCE_INLINE void* taosDecodeSMqHbRsp(void* buf, SMqHbRsp* pRsp) {
buf = taosDecodeFixedI8(buf, &pRsp->status);
buf = taosDecodeFixedI8(buf, &pRsp->vnodeChanged);
buf = taosDecodeFixedI8(buf, &pRsp->epChanged);
buf = taosDecodeSEpSet(buf, &pRsp->epSet);
return buf;
}
typedef struct {
int32_t acctId;
int64_t clusterId;
@ -993,6 +1033,13 @@ typedef struct {
uint64_t taskId;
} SSinkDataReq;
typedef struct {
SMsgHead header;
uint64_t sId;
uint64_t queryId;
uint64_t taskId;
} SQueryContinueReq;
typedef struct {
SMsgHead header;
@ -1053,6 +1100,7 @@ typedef struct {
typedef struct {
int8_t igExists;
char* name;
char* sql;
char* physicalPlan;
char* logicalPlan;
} SCMCreateTopicReq;
@ -1061,6 +1109,7 @@ static FORCE_INLINE int tSerializeSCMCreateTopicReq(void** buf, const SCMCreateT
int tlen = 0;
tlen += taosEncodeFixedI8(buf, pReq->igExists);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeString(buf, pReq->sql);
tlen += taosEncodeString(buf, pReq->physicalPlan);
tlen += taosEncodeString(buf, pReq->logicalPlan);
return tlen;
@ -1069,6 +1118,7 @@ static FORCE_INLINE int tSerializeSCMCreateTopicReq(void** buf, const SCMCreateT
static FORCE_INLINE void* tDeserializeSCMCreateTopicReq(void* buf, SCMCreateTopicReq* pReq) {
buf = taosDecodeFixedI8(buf, &(pReq->igExists));
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeString(buf, &(pReq->sql));
buf = taosDecodeString(buf, &(pReq->physicalPlan));
buf = taosDecodeString(buf, &(pReq->logicalPlan));
return buf;
@ -1191,7 +1241,7 @@ typedef struct {
} SMVSubscribeRsp;
typedef struct {
char name[TSDB_TOPIC_FNAME_LEN];
char name[TSDB_TOPIC_NAME_LEN];
int8_t igExists;
int32_t execLen;
void* executor;

4
include/common/tmsgdef.h
View File

@ -129,7 +129,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_VGROUP_LIST, "mnode-vgroup-list", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "mnode-kill-query", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "mnode-kill-conn", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_HEARTBEAT, "mnode-heartbeat", SClientHbBatchReq, SClientHbBatchRsp)
TD_DEF_MSG_TYPE(TDMT_MND_SHOW, "mnode-show", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_SHOW_RETRIEVE, "mnode-retrieve", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_STATUS, "mnode-status", NULL, NULL)
@ -170,6 +170,8 @@ enum {
TD_DEF_MSG_TYPE(TDMT_VND_DROP_TOPIC, "vnode-drop-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES, "vnode-show-tables", SVShowTablesReq, SVShowTablesRsp)
TD_DEF_MSG_TYPE(TDMT_VND_SHOW_TABLES_FETCH, "vnode-show-tables-fetch", SVShowTablesFetchReq, SVShowTablesFetchRsp)
TD_DEF_MSG_TYPE(TDMT_VND_QUERY_CONTINUE, "vnode-query-continue", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_SCHEDULE_DATA_SINK, "vnode-schedule-data-sink", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)

2
include/common/tname.h
View File

@ -25,14 +25,12 @@
#define T_NAME_ACCT 0x1u
#define T_NAME_DB 0x2u
#define T_NAME_TABLE 0x4u
#define T_NAME_TOPIC 0x8u
typedef struct SName {
uint8_t type; //db_name_t, table_name_t
int32_t acctId;
char dbname[TSDB_DB_NAME_LEN];
char tname[TSDB_TABLE_NAME_LEN];
char topicName[TSDB_TOPIC_NAME_LEN];
} SName;
int32_t tNameExtractFullName(const SName* name, char* dst);

9
include/libs/parser/parsenodes.h
View File

@ -43,15 +43,6 @@ typedef struct SField {
int32_t bytes;
} SField;
typedef struct SParseBasicCtx {
uint64_t requestId;
int32_t acctId;
const char *db;
void *pTransporter;
SEpSet mgmtEpSet;
struct SCatalog *pCatalog;
} SParseBasicCtx;
typedef struct SFieldInfo {
int16_t numOfOutput; // number of column in result
SField *final;

10
include/libs/parser/parser.h
View File

@ -23,11 +23,17 @@ extern "C" {
#include "parsenodes.h"
typedef struct SParseContext {
SParseBasicCtx ctx;
uint64_t requestId;
int32_t acctId;
const char *db;
void *pTransporter;
SEpSet mgmtEpSet;
const char *pSql; // sql string
size_t sqlLen; // length of the sql string
char *pMsg; // extended error message if exists to help identifying the problem in sql statement.
int32_t msgLen; // max length of the msg
struct SCatalog *pCatalog;
} SParseContext;
/**
@ -38,7 +44,7 @@ typedef struct SParseContext {
* @param msg extended error message if exists.
* @return error code
*/
int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQuery);
int32_t qParseQuerySql(SParseContext* pContext, SQueryNode** pQueryNode);
/**
* Return true if it is a ddl/dcl sql statement

24
include/libs/planner/planner.h
View File

@ -91,8 +91,11 @@ typedef struct SPhyNode {
typedef struct SScanPhyNode {
SPhyNode node;
uint64_t uid; // unique id of the table
uint64_t uid; // unique id of the table
int8_t tableType;
int32_t order; // scan order: TSDB_ORDER_ASC|TSDB_ORDER_DESC
int32_t count; // repeat count
int32_t reverse; // reverse scan count
} SScanPhyNode;
typedef SScanPhyNode SSystemTableScanPhyNode;
@ -117,6 +120,25 @@ typedef struct SExchangePhyNode {
SArray *pSrcEndPoints; // SEpAddr, scheduler fill by calling qSetSuplanExecutionNode
} SExchangePhyNode;
typedef enum EAggAlgo {
AGG_ALGO_PLAIN = 1, // simple agg across all input rows
AGG_ALGO_SORTED, // grouped agg, input must be sorted
AGG_ALGO_HASHED // grouped agg, use internal hashtable
} EAggAlgo;
typedef enum EAggSplit {
AGG_SPLIT_PRE = 1, // first level agg, maybe don't need calculate the final result
AGG_SPLIT_FINAL // second level agg, must calculate the final result
} EAggSplit;
typedef struct SAggPhyNode {
SPhyNode node;
EAggAlgo aggAlgo; // algorithm used by agg operator
EAggSplit aggSplit; // distributed splitting mode
SArray *pExprs; // SExprInfo list, these are expression list of group_by_clause and parameter expression of aggregate function
SArray *pGroupByList; // SColIndex list, but these must be column node
} SAggPhyNode;
typedef struct SSubplanId {
uint64_t queryId;
uint64_t templateId;

2
include/libs/planner/plannerOp.h
View File

@ -30,7 +30,7 @@ OP_ENUM_MACRO(TagScan)
OP_ENUM_MACRO(SystemTableScan)
OP_ENUM_MACRO(Aggregate)
OP_ENUM_MACRO(Project)
OP_ENUM_MACRO(Groupby)
// OP_ENUM_MACRO(Groupby)
OP_ENUM_MACRO(Limit)
OP_ENUM_MACRO(SLimit)
OP_ENUM_MACRO(TimeWindow)

19
include/libs/qworker/qworker.h
View File

@ -22,9 +22,18 @@ extern "C" {
#include "trpc.h"
enum {
NODE_TYPE_VNODE = 1,
NODE_TYPE_QNODE,
NODE_TYPE_SNODE,
};
typedef struct SQWorkerCfg {
uint32_t maxSchedulerNum;
uint32_t maxResCacheNum;
uint32_t maxTaskNum;
uint32_t maxSchTaskNum;
} SQWorkerCfg;
@ -39,11 +48,17 @@ typedef struct {
uint64_t numOfErrors;
} SQWorkerStat;
typedef int32_t (*putReqToQueryQFp)(void *, struct SRpcMsg *);
int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt);
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp);
int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessQueryContinueMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessDataSinkMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);

64
include/libs/tkv/tkv.h
View File

@ -1,64 +0,0 @@
/*
* 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_TKV_H_
#define _TD_TKV_H_
#if 0
#include "os.h"
#ifdef __cplusplus
extern "C" {
#endif
// Types exported
typedef struct STkvDb STkvDb;
typedef struct STkvOpts STkvOpts;
typedef struct STkvCache STkvCache;
typedef struct STkvReadOpts STkvReadOpts;
typedef struct STkvWriteOpts STkvWriteOpts;
// DB operations
STkvDb *tkvOpen(const STkvOpts *options, const char *path);
void tkvClose(STkvDb *db);
void tkvPut(STkvDb *db, const STkvWriteOpts *, const char *key, size_t keylen, const char *val, size_t vallen);
char * tkvGet(STkvDb *db, const STkvReadOpts *, const char *key, size_t keylen, size_t *vallen);
void tkvCommit(STkvDb *db);
// DB options
STkvOpts *tkvOptsCreate();
void tkvOptsDestroy(STkvOpts *);
void tkvOptionsSetCache(STkvOpts *, STkvCache *);
void tkvOptsSetCreateIfMissing(STkvOpts *, unsigned char);
// DB cache
typedef enum { TKV_LRU_CACHE = 0, TKV_LFU_CACHE = 1 } ETkvCacheType;
STkvCache *tkvCacheCreate(size_t capacity, ETkvCacheType type);
void tkvCacheDestroy(STkvCache *);
// STkvReadOpts
STkvReadOpts *tkvReadOptsCreate();
void tkvReadOptsDestroy(STkvReadOpts *);
// STkvWriteOpts
STkvWriteOpts *tkvWriteOptsCreate();
void tkvWriteOptsDestroy(STkvWriteOpts *);
#ifdef __cplusplus
}
#endif
#endif
#endif /*_TD_TKV_H_*/

1
include/util/taoserror.h
View File

@ -307,6 +307,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_VND_NO_WRITE_AUTH TAOS_DEF_ERROR_CODE(0, 0x0512) //"Database write operation denied")
#define TSDB_CODE_VND_IS_SYNCING TAOS_DEF_ERROR_CODE(0, 0x0513) //"Database is syncing")
#define TSDB_CODE_VND_INVALID_TSDB_STATE TAOS_DEF_ERROR_CODE(0, 0x0514) //"Invalid tsdb state")
#define TSDB_CODE_VND_TB_NOT_EXIST TAOS_DEF_ERROR_CODE(0, 0x0515) // "Table not exists")
// tsdb
#define TSDB_CODE_TDB_INVALID_TABLE_ID TAOS_DEF_ERROR_CODE(0, 0x0600) //"Invalid table ID")

32
include/util/tarray.h
View File

@ -23,10 +23,23 @@ extern "C" {
#include "os.h"
#include "talgo.h"
#if 0
#define TARRAY(TYPE) \
struct { \
int32_t tarray_size_; \
int32_t tarray_neles_; \
struct TYPE* td_array_data_; \
}
#define TARRAY_SIZE(ARRAY) (ARRAY)->tarray_size_
#define TARRAY_NELES(ARRAY) (ARRAY)->tarray_neles_
#define TARRAY_ELE_AT(ARRAY, IDX) ((ARRAY)->td_array_data_ + idx)
#endif
#define TARRAY_MIN_SIZE 8
#define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
#define TARRAY_GET_START(array) ((array)->pData)
#define TARRAY_ELEM_IDX(array, ele) (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
#define TARRAY_GET_START(array) ((array)->pData)
typedef struct SArray {
size_t size;
@ -57,7 +70,7 @@ int32_t taosArrayEnsureCap(SArray* pArray, size_t tsize);
* @param nEles
* @return
*/
void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
void* taosArrayAddBatch(SArray* pArray, const void* pData, int nEles);
/**
*
@ -65,7 +78,7 @@ void *taosArrayAddBatch(SArray *pArray, const void *pData, int nEles);
* @param pData position array list
* @param numOfElems the number of removed position
*/
void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfElems);
void taosArrayRemoveBatch(SArray* pArray, const int32_t* pData, int32_t numOfElems);
/**
*
@ -73,7 +86,7 @@ void taosArrayRemoveBatch(SArray *pArray, const int32_t* pData, int32_t numOfEle
* @param comparFn
* @param fp
*/
void taosArrayRemoveDuplicate(SArray *pArray, __compar_fn_t comparFn, void (*fp)(void*));
void taosArrayRemoveDuplicate(SArray* pArray, __compar_fn_t comparFn, void (*fp)(void*));
/**
* add all element from the source array list into the destination
@ -242,19 +255,18 @@ int32_t taosArraySearchIdx(const SArray* pArray, const void* key, __compar_fn_t
*/
char* taosArraySearchString(const SArray* pArray, const char* key, __compar_fn_t comparFn, int flags);
/**
* sort the pointer data in the array
* @param pArray
* @param compar
* @param param
* @param compar
* @param param
* @return
*/
void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void *param);
void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param);
#ifdef __cplusplus
}
#endif
#endif /*_TD_UTIL_ARRAY_H*/
#endif /*_TD_UTIL_ARRAY_H*/

1
include/util/tcoding.h
View File

@ -351,6 +351,7 @@ static FORCE_INLINE void *taosDecodeString(void *buf, char **value) {
buf = taosDecodeVariantU64(buf, &size);
*value = (char *)malloc((size_t)size + 1);
if (*value == NULL) return NULL;
memcpy(*value, buf, (size_t)size);

2
include/util/tdef.h
View File

@ -181,7 +181,7 @@ do { \
#define TSDB_COL_NAME_LEN 65
#define TSDB_MAX_SAVED_SQL_LEN TSDB_MAX_COLUMNS * 64
#define TSDB_MAX_SQL_LEN TSDB_PAYLOAD_SIZE
#define TSDB_MAX_SQL_SHOW_LEN 512
#define TSDB_MAX_SQL_SHOW_LEN 1024
#define TSDB_MAX_ALLOWED_SQL_LEN (1*1024*1024u) // sql length should be less than 1mb
#define TSDB_APP_NAME_LEN TSDB_UNI_LEN

78
source/client/inc/clientHb.h
View File

@ -1,78 +0,0 @@
/*
* 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/>.
*/
#include "os.h"
#include "tarray.h"
#include "thash.h"
#include "tmsg.h"
#define HEARTBEAT_INTERVAL 1500 // ms
typedef enum {
HEARTBEAT_TYPE_MQ = 0,
// types can be added here
//
HEARTBEAT_TYPE_MAX
} EHbType;
typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq);
typedef struct SAppHbMgr {
// statistics
int32_t reportCnt;
int32_t connKeyCnt;
int64_t reportBytes; // not implemented
int64_t startTime;
// ctl
SRWLatch lock; // lock is used in serialization
// connection
void* transporter;
SEpSet epSet;
// info
SHashObj* activeInfo; // hash<SClientHbKey, SClientHbReq>
SHashObj* getInfoFuncs; // hash<SClientHbKey, FGetConnInfo>
} SAppHbMgr;
typedef struct SClientHbMgr {
int8_t inited;
// ctl
int8_t threadStop;
pthread_t thread;
pthread_mutex_t lock; // used when app init and cleanup
SArray* appHbMgrs; // SArray<SAppHbMgr*> one for each cluster
FHbRspHandle handle[HEARTBEAT_TYPE_MAX];
} SClientHbMgr;
// TODO: embed param into function
// return type: SArray<Skv>
typedef SArray* (*FGetConnInfo)(SClientHbKey connKey, void* param);
// global, called by mgmt
int hbMgrInit();
void hbMgrCleanUp();
int hbHandleRsp(SClientHbBatchRsp* hbRsp);
// cluster level
SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet);
void appHbMgrCleanup(SAppHbMgr* pAppHbMgr);
// conn level
int hbRegisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, FGetConnInfo func);
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);
// mq
void hbMgrInitMqHbRspHandle();

94
source/client/inc/clientInt.h
View File

@ -31,6 +31,41 @@ extern "C" {
#include "trpc.h"
#include "query.h"
#define HEARTBEAT_INTERVAL 1500 // ms
typedef struct SAppInstInfo SAppInstInfo;
typedef int32_t (*FHbRspHandle)(SClientHbRsp* pReq);
typedef struct SAppHbMgr {
// statistics
int32_t reportCnt;
int32_t connKeyCnt;
int64_t reportBytes; // not implemented
int64_t startTime;
// ctl
SRWLatch lock; // lock is used in serialization
// connection
SAppInstInfo* pAppInstInfo;
// info
SHashObj* activeInfo; // hash<SClientHbKey, SClientHbReq>
SHashObj* getInfoFuncs; // hash<SClientHbKey, FGetConnInfo>
} SAppHbMgr;
typedef struct SClientHbMgr {
int8_t inited;
// ctl
int8_t threadStop;
pthread_t thread;
pthread_mutex_t lock; // used when app init and cleanup
SArray* appHbMgrs; // SArray<SAppHbMgr*> one for each cluster
FHbRspHandle handle[HEARTBEAT_TYPE_MAX];
} SClientHbMgr;
// TODO: embed param into function
// return type: SArray<Skv>
typedef SArray* (*FGetConnInfo)(SClientHbKey connKey, void* param);
typedef struct SQueryExecMetric {
int64_t start; // start timestamp
int64_t parsed; // start to parse
@ -55,15 +90,15 @@ typedef struct SHeartBeatInfo {
void *pTimer; // timer, used to send request msg to mnode
} SHeartBeatInfo;
typedef struct SAppInstInfo {
int64_t numOfConns;
SCorEpSet mgmtEp;
SInstanceSummary summary;
struct SAppInstInfo {
int64_t numOfConns;
SCorEpSet mgmtEp;
SInstanceSummary summary;
SList *pConnList; // STscObj linked list
int64_t clusterId;
int64_t clusterId;
void *pTransporter;
SHeartBeatInfo hb;
} SAppInstInfo;
struct SAppHbMgr *pAppHbMgr;
};
typedef struct SAppInfo {
int64_t startTime;
@ -81,25 +116,28 @@ typedef struct STscObj {
char db[TSDB_DB_FNAME_LEN];
int32_t acctId;
uint32_t connId;
int32_t connType;
uint64_t id; // ref ID returned by taosAddRef
void *pTransporter;
pthread_mutex_t mutex; // used to protect the operation on db
int32_t numOfReqs; // number of sqlObj from this tscObj
int32_t numOfReqs; // number of sqlObj bound to this connection
SAppInstInfo *pAppInfo;
} STscObj;
typedef struct SMqConsumer {
STscObj* pTscObj;
} SMqConsumer;
typedef struct SReqResultInfo {
const char *pRspMsg;
const char *pData;
TAOS_FIELD *fields;
uint32_t numOfCols;
int32_t *length;
TAOS_ROW row;
char **pCol;
uint32_t numOfRows;
uint32_t current;
bool completed;
} SReqResultInfo;
typedef struct SShowReqInfo {
@ -110,12 +148,13 @@ typedef struct SShowReqInfo {
} SShowReqInfo;
typedef struct SRequestSendRecvBody {
tsem_t rspSem; // not used now
void* fp;
SShowReqInfo showInfo; // todo this attribute will be removed after the query framework being completed.
struct SSchJob *pQueryJob; // query job, created according to sql query DAG.
SDataBuf requestMsg;
SReqResultInfo resInfo;
tsem_t rspSem; // not used now
void* fp;
SShowReqInfo showInfo; // todo this attribute will be removed after the query framework being completed.
SDataBuf requestMsg;
struct SSchJob *pQueryJob; // query job, created according to sql query DAG.
struct SQueryDag *pDag; // the query dag, generated according to the sql statement.
SReqResultInfo resInfo;
} SRequestSendRecvBody;
#define ERROR_MSG_BUF_DEFAULT_SIZE 512
@ -130,7 +169,6 @@ typedef struct SRequestObj {
char *msgBuf;
void *pInfo; // sql parse info, generated by parser module
int32_t code;
uint64_t affectedRows; // todo remove it
SQueryExecMetric metric;
SRequestSendRecvBody body;
} SRequestObj;
@ -171,6 +209,26 @@ void *doFetchRow(SRequestObj* pRequest);
void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows);
// --- heartbeat
// global, called by mgmt
int hbMgrInit();
void hbMgrCleanUp();
int hbHandleRsp(SClientHbBatchRsp* hbRsp);
// cluster level
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo);
void appHbMgrCleanup(SAppHbMgr* pAppHbMgr);
// conn level
int hbRegisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, FGetConnInfo func);
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);
// --- mq
void hbMgrInitMqHbRspHandle();
#ifdef __cplusplus
}
#endif

12
source/client/src/clientEnv.c
View File

@ -89,13 +89,12 @@ static void tscInitLogFile() {
// todo close the transporter properly
void closeTransporter(STscObj* pTscObj) {
if (pTscObj == NULL || pTscObj->pTransporter == NULL) {
if (pTscObj == NULL || pTscObj->pAppInfo->pTransporter == NULL) {
return;
}
tscDebug("free transporter:%p in connObj: 0x%"PRIx64, pTscObj->pTransporter, pTscObj->id);
rpcClose(pTscObj->pTransporter);
pTscObj->pTransporter = NULL;
tscDebug("free transporter:%p in connObj: 0x%"PRIx64, pTscObj->pAppInfo->pTransporter, pTscObj->id);
rpcClose(pTscObj->pAppInfo->pTransporter);
}
// TODO refactor
@ -140,10 +139,6 @@ void* createTscObj(const char* user, const char* auth, const char *db, SAppInstI
}
pObj->pAppInfo = pAppInfo;
if (pAppInfo != NULL) {
pObj->pTransporter = pAppInfo->pTransporter;
}
tstrncpy(pObj->user, user, sizeof(pObj->user));
memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);
@ -199,6 +194,7 @@ static void doDestroyRequest(void* p) {
tfree(pRequest->pInfo);
doFreeReqResultInfo(&pRequest->body.resInfo);
qDestroyQueryDag(pRequest->body.pDag);
deregisterRequest(pRequest);
tfree(pRequest);

91
source/client/src/clientHb.c
View File

@ -13,7 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "clientHb.h"
#include "clientInt.h"
#include "trpc.h"
static SClientHbMgr clientHbMgr = {0};
@ -21,10 +21,18 @@ static SClientHbMgr clientHbMgr = {0};
static int32_t hbCreateThread();
static void hbStopThread();
static int32_t hbMqHbRspHandle(SClientHbRsp* pReq) {
static int32_t hbMqHbRspHandle(SClientHbRsp* pRsp) {
return 0;
}
static int32_t hbMqAsyncCallBack(void* param, const SDataBuf* pMsg, int32_t code) {
if (code != 0) {
return -1;
}
SClientHbRsp* pRsp = (SClientHbRsp*) pMsg->pData;
return hbMqHbRspHandle(pRsp);
}
void hbMgrInitMqHbRspHandle() {
clientHbMgr.handle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle;
}
@ -35,18 +43,18 @@ static FORCE_INLINE void hbMgrInitHandle() {
}
SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
SClientHbBatchReq* pReq = malloc(sizeof(SClientHbBatchReq));
if (pReq == NULL) {
SClientHbBatchReq* pBatchReq = malloc(sizeof(SClientHbBatchReq));
if (pBatchReq == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
return NULL;
}
int32_t connKeyCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
pReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq));
pBatchReq->reqs = taosArrayInit(connKeyCnt, sizeof(SClientHbReq));
void *pIter = taosHashIterate(pAppHbMgr->activeInfo, NULL);
while (pIter != NULL) {
taosArrayPush(pReq->reqs, pIter);
SClientHbReq* pOneReq = pIter;
taosArrayPush(pBatchReq->reqs, pOneReq);
taosHashClear(pOneReq->info);
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
@ -59,10 +67,10 @@ SClientHbBatchReq* hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
taosHashCopyKey(pIter, &connKey);
getConnInfoFp(connKey, NULL);
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
pIter = taosHashIterate(pAppHbMgr->getInfoFuncs, pIter);
}
return pReq;
return pBatchReq;
}
static void* hbThreadFunc(void* param) {
@ -75,20 +83,48 @@ static void* hbThreadFunc(void* param) {
int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
for(int i = 0; i < sz; i++) {
SAppHbMgr* pAppHbMgr = taosArrayGet(clientHbMgr.appHbMgrs, i);
SClientHbBatchReq* pReq = hbGatherAllInfo(pAppHbMgr);
void* reqStr = NULL;
int tlen = tSerializeSClientHbBatchReq(&reqStr, pReq);
SMsgSendInfo info;
/*info.fp = hbHandleRsp;*/
SAppHbMgr* pAppHbMgr = taosArrayGetP(clientHbMgr.appHbMgrs, i);
int32_t connCnt = atomic_load_32(&pAppHbMgr->connKeyCnt);
if (connCnt == 0) {
continue;
}
SClientHbBatchReq* pReq = hbGatherAllInfo(pAppHbMgr);
if (pReq == NULL) {
continue;
}
int tlen = tSerializeSClientHbBatchReq(NULL, pReq);
void *buf = malloc(tlen);
if (buf == NULL) {
//TODO: error handling
break;
}
void *bufCopy = buf;
tSerializeSClientHbBatchReq(&bufCopy, pReq);
SMsgSendInfo *pInfo = malloc(sizeof(SMsgSendInfo));
if (pInfo == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
tFreeClientHbBatchReq(pReq);
free(buf);
break;
}
pInfo->fp = hbMqAsyncCallBack;
pInfo->msgInfo.pData = buf;
pInfo->msgInfo.len = tlen;
pInfo->msgType = TDMT_MND_HEARTBEAT;
pInfo->param = NULL;
pInfo->requestId = generateRequestId();
pInfo->requestObjRefId = 0;
SAppInstInfo *pAppInstInfo = pAppHbMgr->pAppInstInfo;
int64_t transporterId = 0;
asyncSendMsgToServer(pAppHbMgr->transporter, &pAppHbMgr->epSet, &transporterId, &info);
SEpSet epSet = getEpSet_s(&pAppInstInfo->mgmtEp);
asyncSendMsgToServer(pAppInstInfo->pTransporter, &epSet, &transporterId, pInfo);
tFreeClientHbBatchReq(pReq);
atomic_add_fetch_32(&pAppHbMgr->reportCnt, 1);
taosMsleep(HEARTBEAT_INTERVAL);
}
taosMsleep(HEARTBEAT_INTERVAL);
}
return NULL;
}
@ -110,7 +146,8 @@ static void hbStopThread() {
atomic_store_8(&clientHbMgr.threadStop, 1);
}
SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet) {
SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo) {
hbMgrInit();
SAppHbMgr* pAppHbMgr = malloc(sizeof(SAppHbMgr));
if (pAppHbMgr == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -119,16 +156,27 @@ SAppHbMgr* appHbMgrInit(void* transporter, SEpSet epSet) {
// init stat
pAppHbMgr->startTime = taosGetTimestampMs();
// init connection info
pAppHbMgr->transporter = transporter;
pAppHbMgr->epSet = epSet;
// init app info
pAppHbMgr->pAppInstInfo = pAppInstInfo;
// init hash info
pAppHbMgr->activeInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
if (pAppHbMgr->activeInfo == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
free(pAppHbMgr);
return NULL;
}
pAppHbMgr->activeInfo->freeFp = tFreeClientHbReq;
// init getInfoFunc
pAppHbMgr->getInfoFuncs = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
if (pAppHbMgr->getInfoFuncs == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
free(pAppHbMgr);
return NULL;
}
taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);
return pAppHbMgr;
}
@ -138,7 +186,7 @@ void appHbMgrCleanup(SAppHbMgr* pAppHbMgr) {
int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
for (int i = 0; i < sz; i++) {
SAppHbMgr* pTarget = taosArrayGet(clientHbMgr.appHbMgrs, i);
SAppHbMgr* pTarget = taosArrayGetP(clientHbMgr.appHbMgrs, i);
if (pAppHbMgr == pTarget) {
taosHashCleanup(pTarget->activeInfo);
taosHashCleanup(pTarget->getInfoFuncs);
@ -171,7 +219,6 @@ void hbMgrCleanUp() {
if (old == 0) return;
taosArrayDestroy(clientHbMgr.appHbMgrs);
}
int hbHandleRsp(SClientHbBatchRsp* hbRsp) {

324
source/client/src/clientImpl.c
View File

@ -13,20 +13,21 @@
#include "tpagedfile.h"
#include "tref.h"
#define CHECK_CODE_GOTO(expr, lable) \
#define CHECK_CODE_GOTO(expr, label) \
do { \
int32_t code = expr; \
if (TSDB_CODE_SUCCESS != code) { \
terrno = code; \
goto lable; \
goto label; \
} \
} while (0)
static int32_t initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet);
static SMsgSendInfo* buildConnectMsg(SRequestObj *pRequest);
static void destroySendMsgInfo(SMsgSendInfo* pMsgBody);
static void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp);
static bool stringLengthCheck(const char* str, size_t maxsize) {
static bool stringLengthCheck(const char* str, size_t maxsize) {
if (str == NULL) {
return false;
}
@ -57,7 +58,7 @@ static char* getClusterKey(const char* user, const char* auth, const char* ip, i
return strdup(key);
}
static STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo);
static STscObj* taosConnectImpl(const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo);
static void setResSchemaInfo(SReqResultInfo* pResInfo, const SDataBlockSchema* pDataBlockSchema);
TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass, const char *auth, const char *db, uint16_t port) {
@ -70,18 +71,18 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass,
return NULL;
}
char tmp[TSDB_DB_NAME_LEN] = {0};
char localDb[TSDB_DB_NAME_LEN] = {0};
if (db != NULL) {
if(!validateDbName(db)) {
terrno = TSDB_CODE_TSC_INVALID_DB_LENGTH;
return NULL;
}
tstrncpy(tmp, db, sizeof(tmp));
strdequote(tmp);
tstrncpy(localDb, db, sizeof(localDb));
strdequote(localDb);
}
char secretEncrypt[32] = {0};
char secretEncrypt[TSDB_PASSWORD_LEN + 1] = {0};
if (auth == NULL) {
if (!validatePassword(pass)) {
terrno = TSDB_CODE_TSC_INVALID_PASS_LENGTH;
@ -110,18 +111,20 @@ TAOS *taos_connect_internal(const char *ip, const char *user, const char *pass,
char* key = getClusterKey(user, secretEncrypt, ip, port);
// TODO: race condition here.
SAppInstInfo** pInst = taosHashGet(appInfo.pInstMap, key, strlen(key));
if (pInst == NULL) {
SAppInstInfo* p = calloc(1, sizeof(struct SAppInstInfo));
p->mgmtEp = epSet;
p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores);
p->pAppHbMgr = appHbMgrInit(p);
taosHashPut(appInfo.pInstMap, key, strlen(key), &p, POINTER_BYTES);
pInst = &p;
}
tfree(key);
return taosConnectImpl(ip, user, &secretEncrypt[0], db, port, NULL, NULL, *pInst);
return taosConnectImpl(user, &secretEncrypt[0], localDb, port, NULL, NULL, *pInst);
}
int32_t buildRequest(STscObj *pTscObj, const char *sql, int sqlLen, SRequestObj** pRequest) {
@ -150,23 +153,26 @@ int32_t parseSql(SRequestObj* pRequest, SQueryNode** pQuery) {
STscObj* pTscObj = pRequest->pTscObj;
SParseContext cxt = {
.ctx = {.requestId = pRequest->requestId, .acctId = pTscObj->acctId, .db = getConnectionDB(pTscObj), .pTransporter = pTscObj->pTransporter},
.pSql = pRequest->sqlstr,
.sqlLen = pRequest->sqlLen,
.pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
.requestId = pRequest->requestId,
.acctId = pTscObj->acctId,
.db = getConnectionDB(pTscObj),
.pSql = pRequest->sqlstr,
.sqlLen = pRequest->sqlLen,
.pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
.pTransporter = pTscObj->pAppInfo->pTransporter,
};
cxt.ctx.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.ctx.pCatalog);
cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
if (code != TSDB_CODE_SUCCESS) {
tfree(cxt.ctx.db);
tfree(cxt.db);
return code;
}
code = qParseQuerySql(&cxt, pQuery);
tfree(cxt.ctx.db);
tfree(cxt.db);
return code;
}
@ -187,10 +193,10 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
pShowReqInfo->pArray = pDcl->pExtension;
}
}
asyncSendMsgToServer(pTscObj->pTransporter, &pDcl->epSet, &transporterId, pSendMsg);
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pDcl->epSet, &transporterId, pSendMsg);
} else {
SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
asyncSendMsgToServer(pTscObj->pTransporter, pEpSet, &transporterId, pSendMsg);
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, pEpSet, &transporterId, pSendMsg);
}
tsem_wait(&pRequest->body.rspSem);
@ -212,6 +218,7 @@ int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag)
SSubplan* pPlan = taosArrayGetP(pa, 0);
SDataBlockSchema* pDataBlockSchema = &(pPlan->pDataSink->schema);
setResSchemaInfo(pResInfo, pDataBlockSchema);
pRequest->type = TDMT_VND_QUERY;
}
@ -228,7 +235,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SDataBlockSchema* pDataBlo
SSchema* pSchema = &pDataBlockSchema->pSchema[i];
pResInfo->fields[i].bytes = pSchema->bytes;
pResInfo->fields[i].type = pSchema->type;
tstrncpy(pResInfo->fields[i].name, pSchema[i].name, tListLen(pResInfo->fields[i].name));
tstrncpy(pResInfo->fields[i].name, pSchema->name, tListLen(pResInfo->fields[i].name));
}
}
@ -236,7 +243,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) {
if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
int32_t code = scheduleExecJob(pRequest->pTscObj->pTransporter, NULL, pDag, &pRequest->body.pQueryJob, &res);
int32_t code = scheduleExecJob(pRequest->pTscObj->pAppInfo->pTransporter, NULL, pDag, &pRequest->body.pQueryJob, &res);
if (code != TSDB_CODE_SUCCESS) {
// handle error and retry
} else {
@ -245,43 +252,165 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) {
}
}
pRequest->affectedRows = res.numOfRows;
return res.code;
pRequest->body.resInfo.numOfRows = res.numOfRows;
pRequest->code = res.code;
return pRequest->code;
}
return scheduleAsyncExecJob(pRequest->pTscObj->pTransporter, NULL, pDag, &pRequest->body.pQueryJob);
return scheduleAsyncExecJob(pRequest->pTscObj->pAppInfo->pTransporter, NULL, pDag, &pRequest->body.pQueryJob);
}
TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sqlLen) {
STscObj* pTscObj = (STscObj*)taos;
SRequestObj* pRequest = NULL;
SQueryNode* pQuery = NULL;
SQueryDag* pDag = NULL;
char *dagStr = NULL;
typedef struct tmq_t tmq_t;
typedef struct SMqClientTopic {
// subscribe info
int32_t sqlLen;
char* sql;
char* topicName;
int64_t topicId;
// statistics
int64_t consumeCnt;
// offset
int64_t committedOffset;
int64_t currentOffset;
//connection info
int32_t vgId;
SEpSet epSet;
} SMqClientTopic;
typedef struct tmq_resp_err_t {
int32_t code;
} tmq_resp_err_t;
typedef struct tmq_topic_vgroup_list_t {
char* topicName;
int32_t vgId;
int64_t committedOffset;
} tmq_topic_vgroup_list_t;
typedef void (tmq_commit_cb(tmq_t*, tmq_resp_err_t, tmq_topic_vgroup_list_t*, void* param));
typedef struct tmq_conf_t{
char* clientId;
char* groupId;
char* ip;
uint16_t port;
tmq_commit_cb* commit_cb;
} tmq_conf_t;
struct tmq_t {
char groupId[256];
char clientId[256];
STscObj* pTscObj;
tmq_commit_cb* commit_cb;
SArray* clientTopics; // SArray<SMqClientTopic>
};
void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb) {
conf->commit_cb = cb;
}
SArray* tmqGetConnInfo(SClientHbKey connKey, void* param) {
tmq_t* pTmq = (void*)param;
SArray* pArray = taosArrayInit(0, sizeof(SKv));
if (pArray == NULL) {
return NULL;
}
SKv kv = {0};
kv.key = malloc(256);
if (kv.key == NULL) {
taosArrayDestroy(pArray);
return NULL;
}
strcpy(kv.key, "groupId");
kv.keyLen = strlen("groupId") + 1;
kv.value = malloc(256);
if (kv.value == NULL) {
free(kv.key);
taosArrayDestroy(pArray);
return NULL;
}
strcpy(kv.value, pTmq->groupId);
kv.valueLen = strlen(pTmq->groupId) + 1;
taosArrayPush(pArray, &kv);
strcpy(kv.key, "clientUid");
kv.keyLen = strlen("clientUid") + 1;
*(uint32_t*)kv.value = pTmq->pTscObj->connId;
kv.valueLen = sizeof(uint32_t);
return NULL;
}
tmq_t* tmqCreateConsumerImpl(TAOS* conn, tmq_conf_t* conf) {
tmq_t* pTmq = malloc(sizeof(tmq_t));
if (pTmq == NULL) {
return NULL;
}
strcpy(pTmq->groupId, conf->groupId);
strcpy(pTmq->clientId, conf->clientId);
pTmq->pTscObj = (STscObj*)conn;
pTmq->pTscObj->connType = HEARTBEAT_TYPE_MQ;
return pTmq;
}
TAOS_RES *taos_create_topic(TAOS* taos, const char* topicName, const char* sql, int sqlLen) {
STscObj *pTscObj = (STscObj*)taos;
SRequestObj *pRequest = NULL;
SQueryNode *pQueryNode = NULL;
char *pStr = NULL;
terrno = TSDB_CODE_SUCCESS;
if (taos == NULL || topicName == NULL || sql == NULL) {
tscError("invalid parameters for creating topic, connObj:%p, topic name:%s, sql:%s", taos, topicName, sql);
terrno = TSDB_CODE_TSC_INVALID_INPUT;
goto _return;
}
if (strlen(topicName) >= TSDB_TOPIC_NAME_LEN) {
tscError("topic name too long, max length:%d", TSDB_TOPIC_NAME_LEN - 1);
terrno = TSDB_CODE_TSC_INVALID_INPUT;
goto _return;
}
if (sqlLen > tsMaxSQLStringLen) {
tscError("sql string exceeds max length:%d", tsMaxSQLStringLen);
terrno = TSDB_CODE_TSC_EXCEED_SQL_LIMIT;
goto _return;
}
tscDebug("start to create topic, %s", topicName);
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
CHECK_CODE_GOTO(parseSql(pRequest, &pQueryNode), _return);
//temporary disabled until planner ready
#if 0
CHECK_CODE_GOTO(parseSql(pRequest, &pQuery), _return);
//TODO: check sql valid
// todo check for invalid sql statement and return with error code
CHECK_CODE_GOTO(qCreateQueryDag(pQuery, &pDag), _return);
CHECK_CODE_GOTO(qCreateQueryDag(pQueryNode, &pRequest->body.pDag, pRequest->requestId), _return);
dagStr = qDagToString(pDag);
if(dagStr == NULL) {
//TODO
pStr = qDagToString(pRequest->body.pDag);
if(pStr == NULL) {
goto _return;
}
#endif
// The topic should be related to a database that the queried table is belonged to.
SName name = {0};
char dbName[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(&((SQueryStmtInfo*) pQueryNode)->pTableMetaInfo[0]->name, dbName);
tNameFromString(&name, dbName, T_NAME_ACCT|T_NAME_DB);
tNameFromString(&name, topicName, T_NAME_TABLE);
char topicFname[TSDB_TOPIC_FNAME_LEN] = {0};
tNameExtractFullName(&name, topicFname);
SCMCreateTopicReq req = {
.name = (char*)name,
.igExists = 0,
/*.physicalPlan = dagStr,*/
.physicalPlan = (char*)sql,
.logicalPlan = "",
.name = (char*) topicFname,
.igExists = 0,
.physicalPlan = (char*) pStr,
.sql = (char*) sql,
.logicalPlan = "no logic plan",
};
int tlen = tSerializeSCMCreateTopicReq(NULL, &req);
@ -289,30 +418,54 @@ TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sq
if(buf == NULL) {
goto _return;
}
void* abuf = buf;
tSerializeSCMCreateTopicReq(&abuf, &req);
/*printf("formatted: %s\n", dagStr);*/
pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen };
pRequest->type = TDMT_MND_CREATE_TOPIC;
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
SEpSet epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
int64_t transporterId = 0;
asyncSendMsgToServer(pTscObj->pTransporter, pEpSet, &transporterId, body);
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
_return:
qDestroyQuery(pQuery);
qDestroyQueryDag(pDag);
destroySendMsgInfo(body);
qDestroyQuery(pQueryNode);
if (body != NULL) {
destroySendMsgInfo(body);
}
if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) {
pRequest->code = terrno;
}
return pRequest;
}
typedef struct tmq_message_t {
int32_t numOfRows;
char* topicName;
TAOS_ROW row[];
} tmq_message_t;
tmq_message_t* tmq_consume_poll(tmq_t* mq, int64_t blocking_time) {
return NULL;
}
tmq_resp_err_t* tmq_commit(tmq_t* mq, void* callback, int32_t async) {
return NULL;
}
void tmq_message_destroy(tmq_message_t* mq_message) {
}
TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
STscObj *pTscObj = (STscObj *)taos;
if (sqlLen > (size_t) tsMaxSQLStringLen) {
@ -324,24 +477,22 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
nPrintTsc("%s", sql)
SRequestObj *pRequest = NULL;
SQueryNode *pQuery = NULL;
SQueryDag *pDag = NULL;
SQueryNode *pQueryNode = NULL;
terrno = TSDB_CODE_SUCCESS;
CHECK_CODE_GOTO(buildRequest(pTscObj, sql, sqlLen, &pRequest), _return);
CHECK_CODE_GOTO(parseSql(pRequest, &pQuery), _return);
CHECK_CODE_GOTO(parseSql(pRequest, &pQueryNode), _return);
if (qIsDdlQuery(pQuery)) {
CHECK_CODE_GOTO(execDdlQuery(pRequest, pQuery), _return);
if (qIsDdlQuery(pQueryNode)) {
CHECK_CODE_GOTO(execDdlQuery(pRequest, pQueryNode), _return);
} else {
CHECK_CODE_GOTO(getPlan(pRequest, pQuery, &pDag), _return);
CHECK_CODE_GOTO(scheduleQuery(pRequest, pDag), _return);
CHECK_CODE_GOTO(getPlan(pRequest, pQueryNode, &pRequest->body.pDag), _return);
CHECK_CODE_GOTO(scheduleQuery(pRequest, pRequest->body.pDag), _return);
pRequest->code = terrno;
}
_return:
qDestroyQuery(pQuery);
qDestroyQueryDag(pDag);
qDestroyQuery(pQueryNode);
if (NULL != pRequest && TSDB_CODE_SUCCESS != terrno) {
pRequest->code = terrno;
}
@ -385,7 +536,7 @@ int initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSe
return 0;
}
STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo) {
STscObj* taosConnectImpl(const char *user, const char *auth, const char *db, uint16_t port, __taos_async_fn_t fp, void *param, SAppInstInfo* pAppInfo) {
STscObj *pTscObj = createTscObj(user, auth, db, pAppInfo);
if (NULL == pTscObj) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@ -402,7 +553,7 @@ STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, con
SMsgSendInfo* body = buildConnectMsg(pRequest);
int64_t transporterId = 0;
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
if (pRequest->code != TSDB_CODE_SUCCESS) {
@ -413,7 +564,7 @@ STscObj* taosConnectImpl(const char *ip, const char *user, const char *auth, con
taos_close(pTscObj);
pTscObj = NULL;
} else {
tscDebug("0x%"PRIx64" connection is opening, connId:%d, dnodeConn:%p, reqId:0x%"PRIx64, pTscObj->id, pTscObj->connId, pTscObj->pTransporter, pRequest->requestId);
tscDebug("0x%"PRIx64" connection is opening, connId:%d, dnodeConn:%p, reqId:0x%"PRIx64, pTscObj->id, pTscObj->connId, pTscObj->pAppInfo->pTransporter, pRequest->requestId);
destroyRequest(pRequest);
}
@ -444,7 +595,9 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj *pRequest) {
STscObj *pObj = pRequest->pTscObj;
char* db = getConnectionDB(pObj);
tstrncpy(pConnect->db, db, sizeof(pConnect->db));
if (db != NULL) {
tstrncpy(pConnect->db, db, sizeof(pConnect->db));
}
tfree(db);
pConnect->pid = htonl(appInfo.pid);
@ -545,11 +698,15 @@ void* doFetchRow(SRequestObj* pRequest) {
if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
if (pRequest->type == TDMT_VND_QUERY) {
pRequest->type = TDMT_VND_FETCH;
// All data has returned to App already, no need to try again
if (pResultInfo->completed) {
return NULL;
}
scheduleFetchRows(pRequest->body.pQueryJob, (void **)&pRequest->body.resInfo.pData);
pResultInfo->current = 0;
if (pResultInfo->numOfRows <= pResultInfo->current) {
setQueryResultByRsp(&pRequest->body.resInfo, (SRetrieveTableRsp*)pRequest->body.resInfo.pData);
if (pResultInfo->numOfRows == 0) {
return NULL;
}
@ -577,7 +734,7 @@ void* doFetchRow(SRequestObj* pRequest) {
int64_t transporterId = 0;
STscObj *pTscObj = pRequest->pTscObj;
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
pRequest->type = TDMT_VND_SHOW_TABLES_FETCH;
@ -587,7 +744,7 @@ void* doFetchRow(SRequestObj* pRequest) {
int64_t transporterId = 0;
STscObj *pTscObj = pRequest->pTscObj;
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
@ -611,12 +768,23 @@ _return:
return pResultInfo->row;
}
static void doPrepareResPtr(SReqResultInfo* pResInfo) {
if (pResInfo->row == NULL) {
pResInfo->row = calloc(pResInfo->numOfCols, POINTER_BYTES);
pResInfo->pCol = calloc(pResInfo->numOfCols, POINTER_BYTES);
pResInfo->length = calloc(pResInfo->numOfCols, sizeof(int32_t));
}
}
void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t numOfCols, int32_t numOfRows) {
assert(numOfCols > 0 && pFields != NULL && pResultInfo != NULL);
if (numOfRows == 0) {
return;
}
// todo check for the failure of malloc
doPrepareResPtr(pResultInfo);
int32_t offset = 0;
for (int32_t i = 0; i < numOfCols; ++i) {
pResultInfo->length[i] = pResultInfo->fields[i].bytes;
@ -629,9 +797,12 @@ void setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32_t
char* getConnectionDB(STscObj* pObj) {
char *p = NULL;
pthread_mutex_lock(&pObj->mutex);
p = strndup(pObj->db, tListLen(pObj->db));
pthread_mutex_unlock(&pObj->mutex);
size_t len = strlen(pObj->db);
if (len > 0) {
p = strndup(pObj->db, tListLen(pObj->db));
}
pthread_mutex_unlock(&pObj->mutex);
return p;
}
@ -642,3 +813,14 @@ void setConnectionDB(STscObj* pTscObj, const char* db) {
pthread_mutex_unlock(&pTscObj->mutex);
}
void setQueryResultByRsp(SReqResultInfo* pResultInfo, const SRetrieveTableRsp* pRsp) {
assert(pResultInfo != NULL && pRsp != NULL);
pResultInfo->pRspMsg = (const char*) pRsp;
pResultInfo->pData = (void*) pRsp->data;
pResultInfo->numOfRows = htonl(pRsp->numOfRows);
pResultInfo->current = 0;
pResultInfo->completed = (pRsp->completed == 1);
setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows);
}

8
source/client/src/clientMain.c
View File

@ -265,7 +265,13 @@ const char *taos_data_type(int type) {
const char *taos_get_client_info() { return version; }
int taos_affected_rows(TAOS_RES *res) {
return ((SRequestObj*)res)->affectedRows;
if (res == NULL) {
return 0;
}
SRequestObj* pRequest = (SRequestObj*) res;
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
return pResInfo->numOfRows;
}
int taos_result_precision(TAOS_RES *res) { return TSDB_TIME_PRECISION_MILLI; }

8
source/client/src/clientMsgHandler.c
View File

@ -71,6 +71,9 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pTscObj->pAppInfo->clusterId = pConnect->clusterId;
atomic_add_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
SClientHbKey connKey = {.connId = pConnect->connId, .hbType = HEARTBEAT_TYPE_QUERY};
hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey, NULL);
// pRequest->body.resInfo.pRspMsg = pMsg->pData;
tscDebug("0x%" PRIx64 " clusterId:%" PRId64 ", totalConn:%" PRId64, pRequest->requestId, pConnect->clusterId,
pTscObj->pAppInfo->numOfConns);
@ -154,9 +157,6 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pResInfo->fields = pFields;
pResInfo->numOfCols = pMetaMsg->numOfColumns;
pResInfo->row = calloc(pResInfo->numOfCols, POINTER_BYTES);
pResInfo->pCol = calloc(pResInfo->numOfCols, POINTER_BYTES);
pResInfo->length = calloc(pResInfo->numOfCols, sizeof(int32_t));
pRequest->body.showInfo.execId = pShow->showId;
@ -382,4 +382,4 @@ void initMsgHandleFp() {
handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = processShowRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = processRetrieveVndRsp;
}
}

328
source/client/test/clientTests.cpp
View File

@ -53,6 +53,7 @@ TEST(testCase, connect_Test) {
if (pConn == NULL) {
printf("failed to connect to server, reason:%s\n", taos_errstr(NULL));
}
sleep(3);
taos_close(pConn);
}
@ -148,27 +149,27 @@ TEST(testCase, connect_Test) {
//}
//
//TEST(testCase, create_db_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create database abc1 vgroups 4");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
// taos_close(pConn);
//TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
//assert(pConn != NULL);
//TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
//if (taos_errno(pRes) != 0) {
//printf("error in create db, reason:%s\n", taos_errstr(pRes));
//}
//TAOS_FIELD* pFields = taos_fetch_fields(pRes);
//ASSERT_TRUE(pFields == NULL);
//int32_t numOfFields = taos_num_fields(pRes);
//ASSERT_EQ(numOfFields, 0);
//taos_free_result(pRes);
//pRes = taos_query(pConn, "create database abc1 vgroups 4");
//if (taos_errno(pRes) != 0) {
//printf("error in create db, reason:%s\n", taos_errstr(pRes));
//}
//taos_close(pConn);
//}
//
//TEST(testCase, create_dnode_Test) {
@ -194,7 +195,7 @@ TEST(testCase, connect_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "drop dnode 2");
// TAOS_RES* pRes = taos_query(pConn, "drop dnode 3");
// if (taos_errno(pRes) != 0) {
// printf("error in drop dnode, reason:%s\n", taos_errstr(pRes));
// }
@ -205,6 +206,11 @@ TEST(testCase, connect_Test) {
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// pRes = taos_query(pConn, "drop dnode 4");
// if (taos_errno(pRes) != 0) {
// printf("error in drop dnode, reason:%s\n", taos_errstr(pRes));
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
@ -227,45 +233,39 @@ TEST(testCase, connect_Test) {
// taos_close(pConn);
//}
//
//// TEST(testCase, drop_db_test) {
//// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
//// assert(pConn != NULL);
////
//// showDB(pConn);
////
//// TAOS_RES* pRes = taos_query(pConn, "drop database abc1");
//// if (taos_errno(pRes) != 0) {
//// printf("failed to drop db, reason:%s\n", taos_errstr(pRes));
//// }
//// taos_free_result(pRes);
////
//// showDB(pConn);
////
//// pRes = taos_query(pConn, "create database abc1");
//// if (taos_errno(pRes) != 0) {
//// printf("create to drop db, reason:%s\n", taos_errstr(pRes));
//// }
//// taos_free_result(pRes);
//// taos_close(pConn);
////}
// TEST(testCase, drop_db_test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// showDB(pConn);
//
// TAOS_RES* pRes = taos_query(pConn, "drop database abc1");
// if (taos_errno(pRes) != 0) {
// printf("failed to drop db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// showDB(pConn);
//
// pRes = taos_query(pConn, "create database abc1");
// if (taos_errno(pRes) != 0) {
// printf("create to drop db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
// taos_close(pConn);
//}
//TEST(testCase, create_stable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
// TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in use db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create stable st1(ts timestamp, k int) tags(a int)");
// pRes = taos_query(pConn, "create table if not exists abc1.st1(ts timestamp, k int) tags(a int)");
// if (taos_errno(pRes) != 0) {
// printf("error in create stable, reason:%s\n", taos_errstr(pRes));
// }
@ -277,22 +277,40 @@ TEST(testCase, connect_Test) {
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, create_table_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
// pRes = taos_query(pConn, "create stable if not exists abc1.`123_$^)` (ts timestamp, `abc` int) tags(a int)");
// if (taos_errno(pRes) != 0) {
// printf("failed to create super table 123_$^), reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create table tm0(ts timestamp, k int)");
// taos_free_result(pRes);
// pRes = taos_query(pConn, "drop stable `123_$^)`");
// if (taos_errno(pRes) != 0) {
// printf("failed to drop super table 123_$^), reason:%s\n", taos_errstr(pRes));
// }
//
// taos_close(pConn);
//}
//
TEST(testCase, create_table_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != NULL);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists tm0(ts timestamp, k int)");
ASSERT_EQ(taos_errno(pRes), 0);
taos_free_result(pRes);
pRes = taos_query(pConn, "create table if not exists tm0(ts timestamp, k blob)");
ASSERT_NE(taos_errno(pRes), 0);
taos_free_result(pRes);
taos_close(pConn);
}
//TEST(testCase, create_ctable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
@ -311,37 +329,37 @@ TEST(testCase, connect_Test) {
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, show_stable_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
TEST(testCase, show_stable_Test) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
assert(pConn != nullptr);
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("failed to use db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "show stables");
// if (taos_errno(pRes) != 0) {
// printf("failed to show stables, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// ASSERT_TRUE(false);
// }
//
// TAOS_ROW pRow = NULL;
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
TAOS_RES* pRes = taos_query(pConn, "show abc1.stables");
if (taos_errno(pRes) != 0) {
printf("failed to show stables, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
TAOS_ROW pRow = NULL;
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
int32_t numOfFields = taos_num_fields(pRes);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
//
//TEST(testCase, show_vgroup_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
@ -499,43 +517,34 @@ TEST(testCase, connect_Test) {
//
// taosHashCleanup(phash);
//}
// TEST(testCase, create_topic_Test) {
//
//TEST(testCase, create_topic_Test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// assert(pConn != NULL);
//
// TAOS_RES* pRes = taos_query(pConn, "create database abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in create db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "use abc1");
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// if (taos_errno(pRes) != 0) {
// printf("error in use db, reason:%s\n", taos_errstr(pRes));
// }
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create stable st1(ts timestamp, k int) tags(a int)");
// if (taos_errno(pRes) != 0) {
// printf("error in create stable, reason:%s\n", taos_errstr(pRes));
// }
//
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// ASSERT_TRUE(pFields == NULL);
// ASSERT_TRUE(pFields == nullptr);
//
// int32_t numOfFields = taos_num_fields(pRes);
// ASSERT_EQ(numOfFields, 0);
//
// taos_free_result(pRes);
//
// char* sql = "select * from st1";
// tmq_create_topic(pConn, "test_topic_1", sql, strlen(sql));
// char* sql = "select * from tu";
// pRes = taos_create_topic(pConn, "test_topic_1", sql, strlen(sql));
// taos_free_result(pRes);
// taos_close(pConn);
//}
//
//TEST(testCase, insert_test) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// ASSERT_EQ(pConn, nullptr);
// ASSERT_NE(pConn, nullptr);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
@ -550,49 +559,82 @@ TEST(testCase, connect_Test) {
// taos_free_result(pRes);
// taos_close(pConn);
//}
//#endif
TEST(testCase, projection_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
// TAOS_RES* pRes = taos_query(pConn, "create database abc1 vgroups 2");
//TEST(testCase, projection_query_tables) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// ASSERT_NE(pConn, nullptr);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
//
//// pRes = taos_query(pConn, "create stable st1 (ts timestamp, k int) tags(a int)");
//// if (taos_errno(pRes) != 0) {
//// printf("failed to create table tu, reason:%s\n", taos_errstr(pRes));
//// }
//// taos_free_result(pRes);
////
//// pRes = taos_query(pConn, "create table tu using st1 tags(1)");
//// if (taos_errno(pRes) != 0) {
//// printf("failed to create table tu, reason:%s\n", taos_errstr(pRes));
//// }
//// taos_free_result(pRes);
////
//// for(int32_t i = 0; i < 100; ++i) {
//// char sql[512] = {0};
//// sprintf(sql, "insert into tu values(now+%da, %d)", i, i);
//// TAOS_RES* p = taos_query(pConn, sql);
//// if (taos_errno(p) != 0) {
//// printf("failed to insert data, reason:%s\n", taos_errstr(p));
//// }
////
//// taos_free_result(p);
//// }
//
// pRes = taos_query(pConn, "select * from tu");
// if (taos_errno(pRes) != 0) {
// printf("failed to use db, reason:%s\n", taos_errstr(pRes));
// printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// return;
// ASSERT_TRUE(false);
// }
// taos_free_result(pRes);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
// pRes = taos_query(pConn, "create table m1 (ts timestamp, k int) tags(a int)");
taos_free_result(pRes);
//
// pRes = taos_query(pConn, "create table tu using m1 tags(1)");
// TAOS_ROW pRow = NULL;
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
// taos_close(pConn);
//}
//TEST(testCase, projection_query_stables) {
// TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
// ASSERT_NE(pConn, nullptr);
//
// TAOS_RES* pRes = taos_query(pConn, "use abc1");
// taos_free_result(pRes);
//
// pRes = taos_query(pConn, "insert into tu values(now, 1)");
// pRes = taos_query(pConn, "select ts,k from m1");
// if (taos_errno(pRes) != 0) {
// printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
// taos_free_result(pRes);
// ASSERT_TRUE(false);
// }
//
// TAOS_ROW pRow = NULL;
// TAOS_FIELD* pFields = taos_fetch_fields(pRes);
// int32_t numOfFields = taos_num_fields(pRes);
//
// char str[512] = {0};
// while ((pRow = taos_fetch_row(pRes)) != NULL) {
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
// }
//
// taos_free_result(pRes);
pRes = taos_query(pConn, "select * from tu");
if (taos_errno(pRes) != 0) {
printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
ASSERT_TRUE(false);
}
TAOS_ROW pRow = NULL;
TAOS_FIELD* pFields = taos_fetch_fields(pRes);
int32_t numOfFields = taos_num_fields(pRes);
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
// taos_close(pConn);
//}

74
source/common/src/tmsg.c
View File

@ -89,7 +89,7 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
int tlen = 0;
tlen += taosEncodeSClientHbKey(buf, &pReq->connKey);
int kvNum = taosHashGetSize(pReq->info);
int32_t kvNum = taosHashGetSize(pReq->info);
tlen += taosEncodeFixedI32(buf, kvNum);
SKv kv;
void* pIter = taosHashIterate(pReq->info, pIter);
@ -104,14 +104,15 @@ int tSerializeSClientHbReq(void **buf, const SClientHbReq *pReq) {
return tlen;
}
void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) {
ASSERT(pReq->info != NULL);
void *tDeserializeSClientHbReq(void *buf, SClientHbReq *pReq) {
buf = taosDecodeSClientHbKey(buf, &pReq->connKey);
// TODO: error handling
int kvNum;
taosDecodeFixedI32(buf, &kvNum);
pReq->info = taosHashInit(kvNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
int32_t kvNum;
buf = taosDecodeFixedI32(buf, &kvNum);
if (pReq->info == NULL) {
pReq->info = taosHashInit(kvNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
}
for(int i = 0; i < kvNum; i++) {
SKv kv;
buf = taosDecodeSKv(buf, &kv);
@ -121,12 +122,69 @@ void *tDeserializeClientHbReq(void *buf, SClientHbReq *pReq) {
return buf;
}
int tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pReq) {
int tSerializeSClientHbRsp(void** buf, const SClientHbRsp* pRsp) {
int tlen = 0;
tlen += taosEncodeSClientHbKey(buf, &pRsp->connKey);
tlen += taosEncodeFixedI32(buf, pRsp->status);
tlen += taosEncodeFixedI32(buf, pRsp->bodyLen);
tlen += taosEncodeBinary(buf, pRsp->body, pRsp->bodyLen);
return tlen;
}
void* tDeserializeSClientHbRsp(void* buf, SClientHbRsp* pRsp) {
buf = taosDecodeSClientHbKey(buf, &pRsp->connKey);
buf = taosDecodeFixedI32(buf, &pRsp->status);
buf = taosDecodeFixedI32(buf, &pRsp->bodyLen);
buf = taosDecodeBinary(buf, &pRsp->body, pRsp->bodyLen);
return buf;
}
int tSerializeSClientHbBatchReq(void** buf, const SClientHbBatchReq* pBatchReq) {
int tlen = 0;
tlen += taosEncodeFixedI64(buf, pBatchReq->reqId);
int32_t reqNum = taosArrayGetSize(pBatchReq->reqs);
tlen += taosEncodeFixedI32(buf, reqNum);
for (int i = 0; i < reqNum; i++) {
SClientHbReq* pReq = taosArrayGet(pBatchReq->reqs, i);
tlen += tSerializeSClientHbReq(buf, pReq);
}
return tlen;
}
void* tDeserializeClientHbBatchReq(void* buf, SClientHbBatchReq* pReq) {
void* tDeserializeSClientHbBatchReq(void* buf, SClientHbBatchReq* pBatchReq) {
buf = taosDecodeFixedI64(buf, &pBatchReq->reqId);
if (pBatchReq->reqs == NULL) {
pBatchReq->reqs = taosArrayInit(0, sizeof(SClientHbReq));
}
int32_t reqNum;
buf = taosDecodeFixedI32(buf, &reqNum);
for (int i = 0; i < reqNum; i++) {
SClientHbReq req = {0};
buf = tDeserializeSClientHbReq(buf, &req);
taosArrayPush(pBatchReq->reqs, &req);
}
return buf;
}
int tSerializeSClientHbBatchRsp(void** buf, const SClientHbBatchRsp* pBatchRsp) {
int tlen = 0;
int32_t sz = taosArrayGetSize(pBatchRsp->rsps);
tlen += taosEncodeFixedI32(buf, sz);
for (int i = 0; i < sz; i++) {
SClientHbRsp* pRsp = taosArrayGet(pBatchRsp->rsps, i);
tlen += tSerializeSClientHbRsp(buf, pRsp);
}
return tlen;
}
void* tDeserializeSClientHbBatchRsp(void* buf, SClientHbBatchRsp* pBatchRsp) {
int32_t sz;
buf = taosDecodeFixedI32(buf, &sz);
pBatchRsp->rsps = taosArrayInit(sz, sizeof(SClientHbRsp));
for (int i = 0; i < sz; i++) {
SClientHbRsp rsp = {0};
buf = tDeserializeSClientHbRsp(buf, &rsp);
taosArrayPush(pBatchRsp->rsps, &rsp);
}
return buf;
}

2
source/dnode/mgmt/impl/src/dndEnv.c
View File

@ -293,7 +293,7 @@ int32_t dndInit(const SDnodeEnvCfg *pCfg) {
if (vnodeInit(&vnodeOpt) != 0) {
dError("failed to init vnode since %s", terrstr());
dndCleanup();
return NULL;
return -1;
}
memcpy(&dndEnv.cfg, pCfg, sizeof(SDnodeEnvCfg));

4
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -350,7 +350,7 @@ typedef struct SMqTopicObj {
// TODO: add cache and change name to id
typedef struct SMqConsumerTopic {
char name[TSDB_TOPIC_FNAME_LEN];
char name[TSDB_TOPIC_NAME_LEN];
SList *vgroups; // SList<int32_t>
} SMqConsumerTopic;
@ -409,7 +409,7 @@ typedef struct SMqVGroupHbObj {
#if 0
typedef struct SCGroupObj {
char name[TSDB_TOPIC_FNAME_LEN];
char name[TSDB_TOPIC_NAME_LEN];
int64_t createTime;
int64_t updateTime;
uint64_t uid;

34
source/dnode/mnode/impl/src/mndProfile.c
View File

@ -258,6 +258,39 @@ static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatReq *pReq) {
}
static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
SMnode *pMnode = pReq->pMnode;
char *batchReqStr = pReq->rpcMsg.pCont;
SClientHbBatchReq batchReq = {0};
tDeserializeSClientHbBatchReq(batchReqStr, &batchReq);
SArray *pArray = batchReq.reqs;
int sz = taosArrayGetSize(pArray);
SClientHbBatchRsp batchRsp = {0};
batchRsp.rsps = taosArrayInit(0, sizeof(SClientHbRsp));
for (int i = 0; i < sz; i++) {
SClientHbReq* pHbReq = taosArrayGet(pArray, i);
if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_QUERY) {
} else if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_MQ) {
SClientHbRsp rsp = {
.status = 0,
.connKey = pHbReq->connKey,
.bodyLen = 0,
.body = NULL
};
taosArrayPush(batchRsp.rsps, &rsp);
}
}
int32_t tlen = tSerializeSClientHbBatchRsp(NULL, &batchRsp);
void* buf = rpcMallocCont(tlen);
void* bufCopy = buf;
tSerializeSClientHbBatchRsp(&bufCopy, &batchRsp);
pReq->contLen = tlen;
pReq->pCont = buf;
return 0;
#if 0
SMnode *pMnode = pReq->pMnode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
@ -327,6 +360,7 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
pReq->contLen = sizeof(SConnectRsp);
pReq->pCont = pRsp;
return 0;
#endif
}
static int32_t mndProcessKillQueryReq(SMnodeMsg *pReq) {

49
source/dnode/mnode/impl/src/mndTopic.c
View File

@ -118,11 +118,13 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT64(pRaw, dataPos, &pTopic->dbUid, TOPIC_DECODE_OVER);
SDB_GET_INT32(pRaw, dataPos, &pTopic->version, TOPIC_DECODE_OVER);
SDB_GET_INT32(pRaw, dataPos, &pTopic->sqlLen, TOPIC_DECODE_OVER);
pTopic->sql = calloc(pTopic->sqlLen + 1, sizeof(char));
SDB_GET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_DECODE_OVER);
SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);
SDB_GET_BINARY(pRaw, dataPos, pTopic->logicalPlan, len, TOPIC_DECODE_OVER);
SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);
SDB_GET_BINARY(pRaw, dataPos, pTopic->physicalPlan, len, TOPIC_DECODE_OVER);
// SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);
// SDB_GET_BINARY(pRaw, dataPos, pTopic->logicalPlan, len, TOPIC_DECODE_OVER);
// SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);
// SDB_GET_BINARY(pRaw, dataPos, pTopic->physicalPlan, len, TOPIC_DECODE_OVER);
SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER)
@ -178,7 +180,7 @@ void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {
static SDbObj *mndAcquireDbByTopic(SMnode *pMnode, char *topicName) {
SName name = {0};
tNameFromString(&name, topicName, T_NAME_ACCT | T_NAME_DB | T_NAME_TOPIC);
tNameFromString(&name, topicName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
char db[TSDB_TABLE_FNAME_LEN] = {0};
tNameGetFullDbName(&name, db);
@ -203,20 +205,24 @@ static SDDropTopicReq *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgroup, SMq
return pDrop;
}
static int32_t mndCheckCreateTopicMsg(SCMCreateTopicReq *pCreate) {
static int32_t mndCheckCreateTopicMsg(SCMCreateTopicReq *creattopReq) {
// deserialize and other stuff
return 0;
}
static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCMCreateTopicReq *pCreate, SDbObj *pDb) {
SMqTopicObj topicObj = {0};
tstrncpy(topicObj.name, pCreate->name, TSDB_TABLE_FNAME_LEN);
tstrncpy(topicObj.name, pCreate->name, TSDB_TOPIC_FNAME_LEN);
tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN);
topicObj.createTime = taosGetTimestampMs();
topicObj.updateTime = topicObj.createTime;
topicObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
topicObj.dbUid = pDb->uid;
topicObj.version = 1;
topicObj.sql = strdup(pCreate->sql);
topicObj.physicalPlan = strdup(pCreate->physicalPlan);
topicObj.logicalPlan = strdup(pCreate->logicalPlan);
topicObj.sqlLen = strlen(pCreate->sql);
SSdbRaw *pTopicRaw = mndTopicActionEncode(&topicObj);
if (pTopicRaw == NULL) return -1;
@ -228,46 +234,47 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCMCreateTopicReq
static int32_t mndProcessCreateTopicMsg(SMnodeMsg *pMsg) {
SMnode *pMnode = pMsg->pMnode;
char *msgStr = pMsg->rpcMsg.pCont;
SCMCreateTopicReq *pCreate;
tDeserializeSCMCreateTopicReq(msgStr, pCreate);
mDebug("topic:%s, start to create", pCreate->name);
SCMCreateTopicReq createTopicReq = {0};
tDeserializeSCMCreateTopicReq(msgStr, &createTopicReq);
if (mndCheckCreateTopicMsg(pCreate) != 0) {
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
mDebug("topic:%s, start to create, sql:%s", createTopicReq.name, createTopicReq.sql);
if (mndCheckCreateTopicMsg(&createTopicReq) != 0) {
mError("topic:%s, failed to create since %s", createTopicReq.name, terrstr());
return -1;
}
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, pCreate->name);
SMqTopicObj *pTopic = mndAcquireTopic(pMnode, createTopicReq.name);
if (pTopic != NULL) {
sdbRelease(pMnode->pSdb, pTopic);
if (pCreate->igExists) {
mDebug("topic:%s, already exist, ignore exist is set", pCreate->name);
if (createTopicReq.igExists) {
mDebug("topic:%s, already exist, ignore exist is set", createTopicReq.name);
return 0;
} else {
terrno = TSDB_CODE_MND_TOPIC_ALREADY_EXIST;
mError("db:%s, failed to create since %s", pCreate->name, terrstr());
mError("db:%s, failed to create since %s", createTopicReq.name, terrstr());
return -1;
}
}
SDbObj *pDb = mndAcquireDbByTopic(pMnode, pCreate->name);
SDbObj *pDb = mndAcquireDbByTopic(pMnode, createTopicReq.name);
if (pDb == NULL) {
terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
mError("topic:%s, failed to create since %s", createTopicReq.name, terrstr());
return -1;
}
int32_t code = mndCreateTopic(pMnode, pMsg, pCreate, pDb);
int32_t code = mndCreateTopic(pMnode, pMsg, &createTopicReq, pDb);
mndReleaseDb(pMnode, pDb);
if (code != 0) {
terrno = code;
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
mError("topic:%s, failed to create since %s", createTopicReq.name, terrstr());
return -1;
}
return TSDB_CODE_MND_ACTION_IN_PROGRESS;
return TSDB_CODE_SUCCESS;
}
static int32_t mndDropTopic(SMnode *pMnode, SMnodeMsg *pMsg, SMqTopicObj *pTopic) { return 0; }

39
source/dnode/mnode/impl/test/profile/profile.cpp
View File

@ -96,6 +96,38 @@ TEST_F(MndTestProfile, 03_ConnectMsg_Show) {
}
TEST_F(MndTestProfile, 04_HeartBeatMsg) {
SClientHbBatchReq batchReq;
batchReq.reqs = taosArrayInit(0, sizeof(SClientHbReq));
SClientHbReq req = {0};
req.connKey = {.connId = 123, .hbType = HEARTBEAT_TYPE_MQ};
req.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
SKv kv;
kv.key = (void*)"abc";
kv.keyLen = 4;
kv.value = (void*)"bcd";
kv.valueLen = 4;
taosHashPut(req.info, kv.key, kv.keyLen, kv.value, kv.valueLen);
taosArrayPush(batchReq.reqs, &req);
int32_t tlen = tSerializeSClientHbBatchReq(NULL, &batchReq);
void* buf = (SClientHbBatchReq*)rpcMallocCont(tlen);
void* bufCopy = buf;
tSerializeSClientHbBatchReq(&bufCopy, &batchReq);
SRpcMsg* pMsg = test.SendReq(TDMT_MND_HEARTBEAT, buf, tlen);
ASSERT_NE(pMsg, nullptr);
ASSERT_EQ(pMsg->code, 0);
char* pRspChar = (char*)pMsg->pCont;
SClientHbBatchRsp rsp = {0};
tDeserializeSClientHbBatchRsp(pRspChar, &rsp);
int sz = taosArrayGetSize(rsp.rsps);
ASSERT_EQ(sz, 1);
SClientHbRsp* pRsp = (SClientHbRsp*) taosArrayGet(rsp.rsps, 0);
EXPECT_EQ(pRsp->connKey.connId, 123);
EXPECT_EQ(pRsp->connKey.hbType, HEARTBEAT_TYPE_MQ);
EXPECT_EQ(pRsp->status, 0);
#if 0
int32_t contLen = sizeof(SHeartBeatReq);
SHeartBeatReq* pReq = (SHeartBeatReq*)rpcMallocCont(contLen);
@ -129,9 +161,12 @@ TEST_F(MndTestProfile, 04_HeartBeatMsg) {
EXPECT_EQ(pRsp->epSet.numOfEps, 1);
EXPECT_EQ(pRsp->epSet.port[0], 9031);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
#endif
}
TEST_F(MndTestProfile, 05_KillConnMsg) {
// temporary remove since kill will use new heartbeat msg
#if 0
{
int32_t contLen = sizeof(SKillConnReq);
@ -190,6 +225,7 @@ TEST_F(MndTestProfile, 05_KillConnMsg) {
connId = pRsp->connId;
}
#endif
}
TEST_F(MndTestProfile, 06_KillConnMsg_InvalidConn) {
@ -204,6 +240,8 @@ TEST_F(MndTestProfile, 06_KillConnMsg_InvalidConn) {
}
TEST_F(MndTestProfile, 07_KillQueryMsg) {
// temporary remove since kill will use new heartbeat msg
#if 0
{
int32_t contLen = sizeof(SKillQueryReq);
@ -252,6 +290,7 @@ TEST_F(MndTestProfile, 07_KillQueryMsg) {
EXPECT_EQ(pRsp->epSet.port[0], 9031);
EXPECT_STREQ(pRsp->epSet.fqdn[0], "localhost");
}
#endif
}
TEST_F(MndTestProfile, 08_KillQueryMsg_InvalidConn) {

37
source/dnode/vnode/CMakeLists.txt
View File

@ -1,4 +1,33 @@
add_subdirectory(meta)
add_subdirectory(tq)
add_subdirectory(tsdb)
add_subdirectory(impl)
aux_source_directory(src/meta META_SRC)
aux_source_directory(src/tq TQ_SRC)
aux_source_directory(src/tsdb TSDB_SRC)
aux_source_directory(src/vnd VND_SRC)
list(APPEND
VNODE_SRC
${META_SRC}
${TQ_SRC}
${TSDB_SRC}
${VND_SRC}
)
add_library(vnode STATIC ${VNODE_SRC})
target_include_directories(
vnode
PUBLIC inc
PRIVATE src/inc
)
target_link_libraries(
vnode
PUBLIC os
PUBLIC util
PUBLIC common
PUBLIC transport
PUBLIC bdb
PUBLIC tfs
PUBLIC wal
PUBLIC qworker
)
if(${BUILD_TEST})
# add_subdirectory(test)
endif(${BUILD_TEST})

24
source/dnode/vnode/impl/CMakeLists.txt
View File

@ -1,24 +0,0 @@
aux_source_directory(src VNODE_SRC)
add_library(vnode STATIC ${VNODE_SRC})
target_include_directories(
vnode
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/vnode"
private "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
vnode
PUBLIC os
PUBLIC transport
PUBLIC meta
PUBLIC tq
PUBLIC tsdb
PUBLIC wal
PUBLIC sync
PUBLIC cjson
PUBLIC qworker
)
# test
if(${BUILD_TEST})
# add_subdirectory(test)
endif(${BUILD_TEST})

16
source/dnode/vnode/impl/src/vnodeFS.c
View File

@ -1,16 +0,0 @@
/*
* 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/>.
*/
#include "vnodeDef.h"

16
source/dnode/vnode/impl/src/vnodeRead.c
View File

@ -1,16 +0,0 @@
/*
* 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/>.
*/
#include "vnodeDef.h"

119
source/dnode/vnode/impl/src/vnodeRequest.c
View File

@ -1,119 +0,0 @@
/*
* 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/>.
*/
#include "vnodeDef.h"
#if 0
static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq);
static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq);
int vnodeBuildReq(void **buf, const SVnodeReq *pReq, tmsg_t type) {
int tsize = 0;
tsize += taosEncodeFixedU64(buf, pReq->ver);
switch (type) {
case TDMT_VND_CREATE_STB:
tsize += vnodeBuildCreateTableReq(buf, &(pReq->ctReq));
break;
case TDMT_VND_SUBMIT:
/* code */
break;
default:
break;
}
/* TODO */
return tsize;
}
void *vnodeParseReq(void *buf, SVnodeReq *pReq, tmsg_t type) {
buf = taosDecodeFixedU64(buf, &(pReq->ver));
switch (type) {
case TDMT_VND_CREATE_STB:
buf = vnodeParseCreateTableReq(buf, &(pReq->ctReq));
break;
default:
break;
}
// TODO
return buf;
}
static int vnodeBuildCreateTableReq(void **buf, const SVCreateTableReq *pReq) {
int tsize = 0;
tsize += taosEncodeString(buf, pReq->name);
tsize += taosEncodeFixedU32(buf, pReq->ttl);
tsize += taosEncodeFixedU32(buf, pReq->keep);
tsize += taosEncodeFixedU8(buf, pReq->type);
switch (pReq->type) {
case META_SUPER_TABLE:
tsize += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
tsize += tdEncodeSchema(buf, pReq->stbCfg.pSchema);
tsize += tdEncodeSchema(buf, pReq->stbCfg.pTagSchema);
break;
case META_CHILD_TABLE:
tsize += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
tsize += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
break;
case META_NORMAL_TABLE:
tsize += tdEncodeSchema(buf, pReq->ntbCfg.pSchema);
break;
default:
break;
}
return tsize;
}
static void *vnodeParseCreateTableReq(void *buf, SVCreateTableReq *pReq) {
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type));
switch (pReq->type) {
case META_SUPER_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
buf = tdDecodeSchema(buf, &(pReq->stbCfg.pSchema));
buf = tdDecodeSchema(buf, &(pReq->stbCfg.pTagSchema));
break;
case META_CHILD_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->ctbCfg.suid));
buf = tdDecodeKVRow(buf, &(pReq->ctbCfg.pTag));
break;
case META_NORMAL_TABLE:
buf = tdDecodeSchema(buf, &(pReq->ntbCfg.pSchema));
break;
default:
break;
}
return buf;
}
int vnodeBuildDropTableReq(void **buf, const SVDropTbReq *pReq) {
// TODO
return 0;
}
void *vnodeParseDropTableReq(void *buf, SVDropTbReq *pReq) {
// TODO
}
#endif

14
source/dnode/vnode/impl/src/vnodeSync.c
View File

@ -1,14 +0,0 @@
/*
* 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/>.
*/

12
source/dnode/vnode/impl/test/CMakeLists.txt
View File

@ -1,12 +0,0 @@
# Vnode API test
add_executable(vnodeApiTests "")
target_sources(vnodeApiTests
PRIVATE
"vnodeApiTests.cpp"
)
target_link_libraries(vnodeApiTests vnode gtest gtest_main)
add_test(
NAME vnode_api_tests
COMMAND ${CMAKE_CURRENT_BINARY_DIR}/vnodeApiTests
)

2
source/dnode/vnode/impl/test/vBenchmarkTest.cpp
View File

@ -1,2 +0,0 @@
// https://stackoverflow.com/questions/8565666/benchmarking-with-googletest
// https://github.com/google/benchmark

285
source/dnode/vnode/impl/test/vnodeApiTests.cpp
View File

@ -1,285 +0,0 @@
/**
* @file vnodeApiTests.cpp
* @author hzcheng (hzcheng@taosdata.com)
* @brief VNODE module API tests
* @version 0.1
* @date 2021-12-13
*
* @copyright Copyright (c) 2021
*
*/
#include <gtest/gtest.h>
#include <iostream>
#include "vnode.h"
static STSchema *vtCreateBasicSchema() {
STSchemaBuilder sb;
STSchema * pSchema = NULL;
tdInitTSchemaBuilder(&sb, 0);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
for (int i = 1; i < 10; i++) {
tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, i, 0);
}
pSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pSchema;
}
static STSchema *vtCreateBasicTagSchema() {
STSchemaBuilder sb;
STSchema * pSchema = NULL;
tdInitTSchemaBuilder(&sb, 0);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
for (int i = 10; i < 12; i++) {
tdAddColToSchema(&sb, TSDB_DATA_TYPE_BINARY, i, 20);
}
pSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pSchema;
}
static SKVRow vtCreateBasicTag() {
SKVRowBuilder rb;
SKVRow pTag;
tdInitKVRowBuilder(&rb);
for (int i = 0; i < 2; i++) {
void *pVal = malloc(sizeof(VarDataLenT) + strlen("foo"));
varDataLen(pVal) = strlen("foo");
memcpy(varDataVal(pVal), "foo", strlen("foo"));
tdAddColToKVRow(&rb, i, TSDB_DATA_TYPE_BINARY, pVal);
free(pVal);
}
pTag = tdGetKVRowFromBuilder(&rb);
tdDestroyKVRowBuilder(&rb);
return pTag;
}
static void vtBuildCreateStbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
SRpcMsg * pMsg;
STSchema *pSchema;
STSchema *pTagSchema;
int zs;
void * pBuf;
pSchema = vtCreateBasicSchema();
pTagSchema = vtCreateBasicTagSchema();
SVnodeReq vCreateSTbReq;
vnodeSetCreateStbReq(&vCreateSTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pSchema, pTagSchema);
zs = vnodeBuildReq(NULL, &vCreateSTbReq, TDMT_VND_CREATE_STB);
pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + zs);
pMsg->msgType = TDMT_VND_CREATE_STB;
pMsg->contLen = zs;
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(SRpcMsg));
pBuf = pMsg->pCont;
vnodeBuildReq(&pBuf, &vCreateSTbReq, TDMT_VND_CREATE_STB);
META_CLEAR_TB_CFG(&vCreateSTbReq);
tdFreeSchema(pSchema);
tdFreeSchema(pTagSchema);
*ppMsg = pMsg;
}
static void vtBuildCreateCtbReq(tb_uid_t suid, char *tbname, SRpcMsg **ppMsg) {
SRpcMsg *pMsg;
int tz;
SKVRow pTag = vtCreateBasicTag();
SVnodeReq vCreateCTbReq;
vnodeSetCreateCtbReq(&vCreateCTbReq, tbname, UINT32_MAX, UINT32_MAX, suid, pTag);
tz = vnodeBuildReq(NULL, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
pMsg = (SRpcMsg *)malloc(sizeof(SRpcMsg) + tz);
pMsg->msgType = TDMT_VND_CREATE_TABLE;
pMsg->contLen = tz;
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
void *pBuf = pMsg->pCont;
vnodeBuildReq(&pBuf, &vCreateCTbReq, TDMT_VND_CREATE_TABLE);
META_CLEAR_TB_CFG(&vCreateCTbReq);
free(pTag);
*ppMsg = pMsg;
}
static void vtBuildCreateNtbReq(char *tbname, SRpcMsg **ppMsg) {
// TODO
}
static void vtBuildSubmitReq(SRpcMsg **ppMsg) {
SRpcMsg * pMsg;
SSubmitMsg *pSubmitMsg;
SSubmitBlk *pSubmitBlk;
int tz = 1024; // TODO
pMsg = (SRpcMsg *)malloc(sizeof(*pMsg) + tz);
pMsg->msgType = TDMT_VND_SUBMIT;
pMsg->contLen = tz;
pMsg->pCont = POINTER_SHIFT(pMsg, sizeof(*pMsg));
// For submit msg header
pSubmitMsg = (SSubmitMsg *)(pMsg->pCont);
// pSubmitMsg->header.contLen = 0;
// pSubmitMsg->header.vgId = 0;
// pSubmitMsg->length = 0;
pSubmitMsg->numOfBlocks = 1;
// For submit blk
pSubmitBlk = (SSubmitBlk *)(pSubmitMsg->blocks);
pSubmitBlk->uid = 0;
pSubmitBlk->tid = 0;
pSubmitBlk->padding = 0;
pSubmitBlk->sversion = 0;
pSubmitBlk->dataLen = 0;
pSubmitBlk->numOfRows = 0;
// For row batch
*ppMsg = pMsg;
}
static void vtClearMsgBatch(SArray *pMsgArr) {
SRpcMsg *pMsg;
for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
pMsg = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
free(pMsg);
}
taosArrayClear(pMsgArr);
}
static void vtProcessAndApplyReqs(SVnode *pVnode, SArray *pMsgArr) {
int rcode;
SRpcMsg *pReq;
SRpcMsg *pRsp;
rcode = vnodeProcessWMsgs(pVnode, pMsgArr);
GTEST_ASSERT_EQ(rcode, 0);
for (size_t i = 0; i < taosArrayGetSize(pMsgArr); i++) {
pReq = *(SRpcMsg **)taosArrayGet(pMsgArr, i);
rcode = vnodeApplyWMsg(pVnode, pReq, NULL);
GTEST_ASSERT_EQ(rcode, 0);
}
}
TEST(vnodeApiTest, vnode_simple_create_table_test) {
tb_uid_t suid = 1638166374163;
SRpcMsg *pMsg;
SArray * pMsgArr = NULL;
SVnode * pVnode;
int rcode;
int ntables = 1000000;
int batch = 10;
char tbname[128];
pMsgArr = (SArray *)taosArrayInit(batch, sizeof(pMsg));
vnodeDestroy("vnode1");
GTEST_ASSERT_GE(vnodeInit(2), 0);
// CREATE AND OPEN A VNODE
pVnode = vnodeOpen("vnode1", NULL);
ASSERT_NE(pVnode, nullptr);
// CREATE A SUPER TABLE
sprintf(tbname, "st");
vtBuildCreateStbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
// CREATE A LOT OF CHILD TABLES
for (int i = 0; i < ntables / batch; i++) {
// Build request batch
for (int j = 0; j < batch; j++) {
sprintf(tbname, "ct%d", i * batch + j + 1);
vtBuildCreateCtbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
}
// Process request batch
vtProcessAndApplyReqs(pVnode, pMsgArr);
// Clear request batch
vtClearMsgBatch(pMsgArr);
}
// CLOSE THE VNODE
vnodeClose(pVnode);
vnodeCleanup();
taosArrayDestroy(pMsgArr);
}
TEST(vnodeApiTest, vnode_simple_insert_test) {
const char *vname = "vnode2";
char tbname[128];
tb_uid_t suid = 1638166374163;
SRpcMsg * pMsg;
SArray * pMsgArr;
int rcode;
SVnode * pVnode;
int batch = 1;
int loop = 1000000;
pMsgArr = (SArray *)taosArrayInit(0, sizeof(pMsg));
vnodeDestroy(vname);
GTEST_ASSERT_GE(vnodeInit(2), 0);
// Open a vnode
pVnode = vnodeOpen(vname, NULL);
GTEST_ASSERT_NE(pVnode, nullptr);
// 1. CREATE A SUPER TABLE
sprintf(tbname, "st");
vtBuildCreateStbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
// 2. CREATE A CHILD TABLE
sprintf(tbname, "t0");
vtBuildCreateCtbReq(suid, tbname, &pMsg);
taosArrayPush(pMsgArr, &pMsg);
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
// 3. WRITE A LOT OF TIME-SERIES DATA
for (int j = 0; j < loop; j++) {
for (int i = 0; i < batch; i++) {
vtBuildSubmitReq(&pMsg);
taosArrayPush(pMsgArr, &pMsg);
}
vtProcessAndApplyReqs(pVnode, pMsgArr);
vtClearMsgBatch(pMsgArr);
}
// Close the vnode
vnodeClose(pVnode);
vnodeCleanup();
taosArrayDestroy(pMsgArr);
}

0
include/dnode/vnode/meta/meta.h → source/dnode/vnode/inc/meta.h
View File

0
include/dnode/vnode/tq/tq.h → source/dnode/vnode/inc/tq.h
View File

0
include/dnode/vnode/tsdb/tsdb.h → source/dnode/vnode/inc/tsdb.h
View File

44
include/dnode/vnode/vnode.h → source/dnode/vnode/inc/vnode.h
View File

@ -35,46 +35,26 @@ typedef struct SDnode SDnode;
typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
typedef struct SVnodeCfg {
int32_t vgId;
SDnode *pDnode;
/** vnode buffer pool options */
struct {
/** write buffer size */
uint64_t wsize;
uint64_t ssize;
uint64_t lsize;
/** use heap allocator or arena allocator */
bool isHeapAllocator;
};
/** time to live of tables in this vnode */
int32_t vgId;
SDnode * pDnode;
uint64_t wsize;
uint64_t ssize;
uint64_t lsize;
bool isHeapAllocator;
uint32_t ttl;
/** data to keep in this vnode */
uint32_t keep;
/** if TS data is eventually consistency */
bool isWeak;
/** TSDB config */
bool isWeak;
STsdbCfg tsdbCfg;
/** META config */
SMetaCfg metaCfg;
/** TQ config */
STqCfg tqCfg;
/** WAL config */
SWalCfg walCfg;
STqCfg tqCfg;
SWalCfg walCfg;
} SVnodeCfg;
typedef struct {
int32_t sver;
char *timezone;
char *locale;
char *charset;
char * timezone;
char * locale;
char * charset;
uint16_t nthreads; // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
PutReqToVQueryQFp putReqToVQueryQFp;
} SVnodeOpt;

46
source/dnode/vnode/meta/CMakeLists.txt
View File

@ -1,46 +0,0 @@
set(META_DB_IMPL_LIST "BDB" "SQLITE")
set(META_DB_IMPL "BDB" CACHE STRING "Use BDB as the default META implementation")
set_property(CACHE META_DB_IMPL PROPERTY STRINGS ${META_DB_IMPL_LIST})
if(META_DB_IMPL IN_LIST META_DB_IMPL_LIST)
message(STATUS "META DB Impl: ${META_DB_IMPL}==============")
else()
message(FATAL_ERROR "Invalid META DB IMPL: ${META_DB_IMPL}==============")
endif()
aux_source_directory(src META_SRC)
if(${META_DB_IMPL} STREQUAL "BDB")
list(REMOVE_ITEM META_SRC "src/metaSQLiteImpl.c")
elseif(${META_DB_IMPL} STREQUAL "SQLITE")
list(REMOVE_ITEM META_SRC "src/metaBDBImpl.c")
endif()
add_library(meta STATIC ${META_SRC})
target_include_directories(
meta
PUBLIC "${CMAKE_SOURCE_DIR}/include/dnode/vnode/meta"
PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/index"
PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
)
target_link_libraries(
meta
PUBLIC common
PUBLIC index
)
if(${META_DB_IMPL} STREQUAL "BDB")
target_link_libraries(
meta
PUBLIC bdb
)
elseif(${META_DB_IMPL} STREQUAL "SQLITE")
target_link_libraries(
meta
PUBLIC sqlite
)
endif()
if(${BUILD_TEST})
add_subdirectory(test)
endif(${BUILD_TEST})

212
source/dnode/vnode/meta/src/metaSQLiteImpl.c
View File

@ -1,212 +0,0 @@
/*
* 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/>.
*/
#include "metaDef.h"
#include "sqlite3.h"
struct SMetaDB {
sqlite3 *pDB;
};
int metaOpenDB(SMeta *pMeta) {
char dir[128];
int rc;
char *err = NULL;
pMeta->pDB = (SMetaDB *)calloc(1, sizeof(SMetaDB));
if (pMeta->pDB == NULL) {
// TODO: handle error
return -1;
}
sprintf(dir, "%s/meta.db", pMeta->path);
rc = sqlite3_open(dir, &(pMeta->pDB->pDB));
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to open meta.db\n");
}
// For all tables
rc = sqlite3_exec(pMeta->pDB->pDB,
"CREATE TABLE IF NOT EXISTS tb ("
" tbname VARCHAR(256) NOT NULL UNIQUE,"
" tb_uid INTEGER NOT NULL UNIQUE "
");",
NULL, NULL, &err);
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to create meta table tb since %s\n", err);
}
// For super tables
rc = sqlite3_exec(pMeta->pDB->pDB,
"CREATE TABLE IF NOT EXISTS stb ("
" tb_uid INTEGER NOT NULL UNIQUE,"
" tbname VARCHAR(256) NOT NULL UNIQUE,"
" tb_schema BLOB NOT NULL,"
" tag_schema BLOB NOT NULL"
");",
NULL, NULL, &err);
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to create meta table stb since %s\n", err);
}
// For normal tables
rc = sqlite3_exec(pMeta->pDB->pDB,
"CREATE TABLE IF NOT EXISTS ntb ("
" tb_uid INTEGER NOT NULL UNIQUE,"
" tbname VARCHAR(256) NOT NULL,"
" tb_schema BLOB NOT NULL"
");",
NULL, NULL, &err);
if (rc != SQLITE_OK) {
// TODO: handle error
printf("failed to create meta table ntb since %s\n", err);
}
sqlite3_exec(pMeta->pDB->pDB, "BEGIN;", NULL, NULL, &err);
tfree(err);
return 0;
}
void metaCloseDB(SMeta *pMeta) {
if (pMeta->pDB) {
sqlite3_exec(pMeta->pDB->pDB, "COMMIT;", NULL, NULL, NULL);
sqlite3_close(pMeta->pDB->pDB);
free(pMeta->pDB);
pMeta->pDB = NULL;
}
// TODO
}
int metaSaveTableToDB(SMeta *pMeta, const STbCfg *pTbCfg) {
char sql[256];
char * err = NULL;
int rc;
tb_uid_t uid;
sqlite3_stmt *stmt;
char buf[256];
void * pBuf;
switch (pTbCfg->type) {
case META_SUPER_TABLE:
uid = pTbCfg->stbCfg.suid;
sprintf(sql,
"INSERT INTO tb VALUES (\'%s\', %" PRIu64
");"
"CREATE TABLE IF NOT EXISTS stb_%" PRIu64
" ("
" tb_uid INTEGER NOT NULL UNIQUE,"
" tbname VARCHAR(256),"
" tag1 INTEGER);",
pTbCfg->name, uid, uid);
rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
if (rc != SQLITE_OK) {
printf("failed to create normal table since %s\n", err);
}
sprintf(sql, "INSERT INTO stb VALUES (%" PRIu64 ", %s, ?, ?)", uid, pTbCfg->name);
sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
pBuf = buf;
tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
sqlite3_bind_blob(stmt, 1, buf, POINTER_DISTANCE(pBuf, buf), NULL);
pBuf = buf;
tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
sqlite3_bind_blob(stmt, 2, buf, POINTER_DISTANCE(pBuf, buf), NULL);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
#if 0
sprintf(sql,
"INSERT INTO tb VALUES (?, ?);"
// "INSERT INTO stb VALUES (?, ?, ?, ?);"
// "CREATE TABLE IF NOT EXISTS stb_%" PRIu64
// " ("
// " tb_uid INTEGER NOT NULL UNIQUE,"
// " tbname VARCHAR(256),"
// " tag1 INTEGER);"
,
uid);
rc = sqlite3_prepare_v2(pMeta->pDB->pDB, sql, -1, &stmt, NULL);
if (rc != SQLITE_OK) {
return -1;
}
sqlite3_bind_text(stmt, 1, pTbCfg->name, -1, SQLITE_TRANSIENT);
sqlite3_bind_int64(stmt, 2, uid);
sqlite3_step(stmt);
sqlite3_finalize(stmt);
// sqlite3_bind_int64(stmt, 3, uid);
// sqlite3_bind_text(stmt, 4, pTbCfg->name, -1, SQLITE_TRANSIENT);
// pBuf = buf;
// tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pSchema);
// sqlite3_bind_blob(stmt, 5, buf, POINTER_DISTANCE(pBuf, buf), NULL);
// pBuf = buf;
// tdEncodeSchema(&pBuf, pTbCfg->stbCfg.pTagSchema);
// sqlite3_bind_blob(stmt, 6, buf, POINTER_DISTANCE(pBuf, buf), NULL);
rc = sqliteVjj3_step(stmt);
if (rc != SQLITE_OK) {
printf("failed to create normal table since %s\n", sqlite3_errmsg(pMeta->pDB->pDB));
}
sqlite3_finalize(stmt);
#endif
break;
case META_NORMAL_TABLE:
// uid = metaGenerateUid(pMeta);
// sprintf(sql,
// "INSERT INTO tb VALUES (\'%s\', %" PRIu64
// ");"
// "INSERT INTO ntb VALUES (%" PRIu64 ", \'%s\', );",
// pTbCfg->name, uid, uid, pTbCfg->name, );
// rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
// if (rc != SQLITE_OK) {
// printf("failed to create normal table since %s\n", err);
// }
break;
case META_CHILD_TABLE:
#if 0
uid = metaGenerateUid(pMeta);
// sprintf(sql, "INSERT INTO tb VALUES (\'%s\', %" PRIu64
// ");"
// "INSERT INTO stb_%" PRIu64 " VALUES (%" PRIu64 ", \'%s\', );");
rc = sqlite3_exec(pMeta->pDB->pDB, sql, NULL, NULL, &err);
if (rc != SQLITE_OK) {
printf("failed to create child table since %s\n", err);
}
#endif
break;
default:
break;
}
tfree(err);
return 0;
}
int metaRemoveTableFromDb(SMeta *pMeta, tb_uid_t uid) {
/* TODO */
return 0;
}

24
source/dnode/vnode/meta/test/CMakeLists.txt
View File

@ -1,24 +0,0 @@
# add_executable(metaTest "")
# target_sources(metaTest
# PRIVATE
# "../src/metaMain.c"
# "../src/metaUid.c"
# "metaTests.cpp"
# )
# target_include_directories(metaTest
# PUBLIC
# "${CMAKE_SOURCE_DIR}/include/server/vnode/meta"
# "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
# )
# target_link_libraries(metaTest
# os
# util
# common
# gtest_main
# tkv
# )
# enable_testing()
# add_test(
# NAME meta_test
# COMMAND metaTest
# )

105
source/dnode/vnode/meta/test/metaAPITest.cpp
View File

@ -1,105 +0,0 @@
#if 0
#include <gtest/gtest.h>
#include <string.h>
#include <iostream>
#include "meta.h"
static STSchema *metaGetSimpleSchema() {
STSchema * pSchema = NULL;
STSchemaBuilder sb = {0};
tdInitTSchemaBuilder(&sb, 0);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_TIMESTAMP, 0, 8);
tdAddColToSchema(&sb, TSDB_DATA_TYPE_INT, 1, 4);
pSchema = tdGetSchemaFromBuilder(&sb);
tdDestroyTSchemaBuilder(&sb);
return pSchema;
}
static SKVRow metaGetSimpleTags() {
SKVRowBuilder kvrb = {0};
SKVRow row;
tdInitKVRowBuilder(&kvrb);
int64_t ts = 1634287978000;
int32_t a = 10;
tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_TIMESTAMP, (void *)(&ts));
tdAddColToKVRow(&kvrb, 0, TSDB_DATA_TYPE_INT, (void *)(&a));
row = tdGetKVRowFromBuilder(&kvrb);
tdDestroyKVRowBuilder(&kvrb);
return row;
}
TEST(MetaTest, DISABLED_meta_create_1m_normal_tables_test) {
// Open Meta
SMeta *meta = metaOpen(NULL, NULL);
std::cout << "Meta is opened!" << std::endl;
// Create 1000000 normal tables
META_TABLE_OPTS_DECLARE(tbOpts);
STSchema *pSchema = metaGetSimpleSchema();
char tbname[128];
for (size_t i = 0; i < 1000000; i++) {
sprintf(tbname, "ntb%ld", i);
metaNormalTableOptsInit(&tbOpts, tbname, pSchema);
metaCreateTable(meta, &tbOpts);
metaTableOptsClear(&tbOpts);
}
tdFreeSchema(pSchema);
// Close Meta
metaClose(meta);
std::cout << "Meta is closed!" << std::endl;
// Destroy Meta
metaDestroy("meta");
std::cout << "Meta is destroyed!" << std::endl;
}
TEST(MetaTest, meta_create_1m_child_tables_test) {
// Open Meta
SMeta *meta = metaOpen(NULL);
std::cout << "Meta is opened!" << std::endl;
// Create a super tables
tb_uid_t uid = 477529885843758ul;
META_TABLE_OPTS_DECLARE(tbOpts);
STSchema *pSchema = metaGetSimpleSchema();
STSchema *pTagSchema = metaGetSimpleSchema();
metaSuperTableOptsInit(&tbOpts, "st", uid, pSchema, pTagSchema);
metaCreateTable(meta, &tbOpts);
metaTableOptsClear(&tbOpts);
tdFreeSchema(pSchema);
tdFreeSchema(pTagSchema);
// Create 1000000 child tables
char name[128];
SKVRow row = metaGetSimpleTags();
for (size_t i = 0; i < 1000000; i++) {
sprintf(name, "ctb%ld", i);
metaChildTableOptsInit(&tbOpts, name, uid, row);
metaCreateTable(meta, &tbOpts);
metaTableOptsClear(&tbOpts);
}
kvRowFree(row);
// Close Meta
metaClose(meta);
std::cout << "Meta is closed!" << std::endl;
// Destroy Meta
metaDestroy("meta");
std::cout << "Meta is destroyed!" << std::endl;
}
#endif

0
source/dnode/vnode/meta/inc/metaCache.h → source/dnode/vnode/src/inc/metaCache.h
View File

0
source/dnode/vnode/meta/inc/metaCfg.h → source/dnode/vnode/src/inc/metaCfg.h
View File

0
source/dnode/vnode/meta/inc/metaDB.h → source/dnode/vnode/src/inc/metaDB.h
View File

0
source/dnode/vnode/meta/inc/metaDef.h → source/dnode/vnode/src/inc/metaDef.h
View File

0
source/dnode/vnode/meta/inc/metaIdx.h → source/dnode/vnode/src/inc/metaIdx.h
View File

0
source/dnode/vnode/meta/inc/metaQuery.h → source/dnode/vnode/src/inc/metaQuery.h
View File

0
source/dnode/vnode/meta/inc/metaTbCfg.h → source/dnode/vnode/src/inc/metaTbCfg.h
View File

0
source/dnode/vnode/meta/inc/metaTbTag.h → source/dnode/vnode/src/inc/metaTbTag.h
View File

0
source/dnode/vnode/meta/inc/metaTbUid.h → source/dnode/vnode/src/inc/metaTbUid.h
View File

0
source/dnode/vnode/tq/inc/tqCommit.h → source/dnode/vnode/src/inc/tqCommit.h
View File

0
source/dnode/vnode/tq/inc/tqInt.h → source/dnode/vnode/src/inc/tqInt.h
View File

0
source/dnode/vnode/tq/inc/tqMetaStore.h → source/dnode/vnode/src/inc/tqMetaStore.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbCommit.h → source/dnode/vnode/src/inc/tsdbCommit.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbCompact.h → source/dnode/vnode/src/inc/tsdbCompact.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbDef.h → source/dnode/vnode/src/inc/tsdbDef.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbFS.h → source/dnode/vnode/src/inc/tsdbFS.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbFile.h → source/dnode/vnode/src/inc/tsdbFile.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbHealth.h → source/dnode/vnode/src/inc/tsdbHealth.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbLog.h → source/dnode/vnode/src/inc/tsdbLog.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbMemTable.h → source/dnode/vnode/src/inc/tsdbMemTable.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbMemory.h → source/dnode/vnode/src/inc/tsdbMemory.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbOptions.h → source/dnode/vnode/src/inc/tsdbOptions.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbReadImpl.h → source/dnode/vnode/src/inc/tsdbReadImpl.h
View File

0
source/dnode/vnode/tsdb/inc/tsdbRowMergeBuf.h → source/dnode/vnode/src/inc/tsdbRowMergeBuf.h
View File

0
source/dnode/vnode/impl/inc/vnodeBufferPool.h → source/dnode/vnode/src/inc/vnodeBufferPool.h
View File

0
source/dnode/vnode/impl/inc/vnodeCfg.h → source/dnode/vnode/src/inc/vnodeCfg.h
View File

0
source/dnode/vnode/impl/inc/vnodeCommit.h → source/dnode/vnode/src/inc/vnodeCommit.h
View File

17
source/dnode/vnode/impl/inc/vnodeDef.h → source/dnode/vnode/src/inc/vnodeDef.h
View File

@ -17,7 +17,7 @@
#define _TD_VNODE_DEF_H_
#include "mallocator.h"
#include "sync.h"
// #include "sync.h"
#include "tcoding.h"
#include "tlist.h"
#include "tlockfree.h"
@ -30,12 +30,9 @@
#include "vnodeBufferPool.h"
#include "vnodeCfg.h"
#include "vnodeCommit.h"
#include "vnodeFS.h"
#include "vnodeMemAllocator.h"
#include "vnodeQuery.h"
#include "vnodeRequest.h"
#include "vnodeStateMgr.h"
#include "vnodeSync.h"
#ifdef __cplusplus
extern "C" {
@ -73,8 +70,6 @@ struct SVnode {
STsdb* pTsdb;
STQ* pTq;
SWal* pWal;
SVnodeSync* pSync;
SVnodeFS* pFs;
tsem_t canCommit;
SQHandle* pQuery;
SDnode* pDnode;
@ -84,6 +79,16 @@ int vnodeScheduleTask(SVnodeTask* task);
int32_t vnodePutReqToVQueryQ(SVnode *pVnode, struct SRpcMsg *pReq);
// For Log
extern int32_t vDebugFlag;
#define vFatal(...) do { if (vDebugFlag & DEBUG_FATAL) { taosPrintLog("VND FATAL ", 255, __VA_ARGS__); }} while(0)
#define vError(...) do { if (vDebugFlag & DEBUG_ERROR) { taosPrintLog("VND ERROR ", 255, __VA_ARGS__); }} while(0)
#define vWarn(...) do { if (vDebugFlag & DEBUG_WARN) { taosPrintLog("VND WARN ", 255, __VA_ARGS__); }} while(0)
#define vInfo(...) do { if (vDebugFlag & DEBUG_INFO) { taosPrintLog("VND ", 255, __VA_ARGS__); }} while(0)
#define vDebug(...) do { if (vDebugFlag & DEBUG_DEBUG) { taosPrintLog("VND ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#define vTrace(...) do { if (vDebugFlag & DEBUG_TRACE) { taosPrintLog("VND ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#ifdef __cplusplus
}
#endif

2
source/dnode/vnode/impl/inc/vnodeInt.h → source/dnode/vnode/src/inc/vnodeInt.h
View File

@ -19,7 +19,7 @@
#include "vnode.h"
#include "meta.h"
#include "sync.h"
// #include "sync.h"
#include "tlog.h"
#include "tq.h"
#include "tsdb.h"

0
source/dnode/vnode/impl/inc/vnodeMemAllocator.h → source/dnode/vnode/src/inc/vnodeMemAllocator.h
View File

4
source/dnode/vnode/impl/inc/vnodeQuery.h → source/dnode/vnode/src/inc/vnodeQuery.h
View File

@ -19,12 +19,12 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "vnodeInt.h"
#include "qworker.h"
#include "vnode.h"
typedef struct SQWorkerMgmt SQHandle;
int vnodeQueryOpen(SVnode *pVnode);
#ifdef __cplusplus

0
source/dnode/vnode/impl/inc/vnodeStateMgr.h → source/dnode/vnode/src/inc/vnodeStateMgr.h
View File

0
source/dnode/vnode/impl/inc/vnodeWrite.h → source/dnode/vnode/src/inc/vnodeWrite.h
View File

0
source/dnode/vnode/meta/src/metaBDBImpl.c → source/dnode/vnode/src/meta/metaBDBImpl.c
View File

0
source/dnode/vnode/meta/src/metaCache.c → source/dnode/vnode/src/meta/metaCache.c
View File

0
source/dnode/vnode/meta/src/metaCfg.c → source/dnode/vnode/src/meta/metaCfg.c
View File

0
source/dnode/vnode/meta/src/metaCommit.c → source/dnode/vnode/src/meta/metaCommit.c
View File

2
source/dnode/vnode/meta/src/metaIdx.c → source/dnode/vnode/src/meta/metaIdx.c
View File

@ -13,7 +13,9 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifdef USE_INVERTED_INDEX
#include "index.h"
#endif
#include "metaDef.h"
struct SMetaIdx {

0
source/dnode/vnode/meta/src/metaMain.c → source/dnode/vnode/src/meta/metaMain.c
View File

0
source/dnode/vnode/meta/src/metaQuery.c → source/dnode/vnode/src/meta/metaQuery.c
View File

0
source/dnode/vnode/meta/src/metaTable.c → source/dnode/vnode/src/meta/metaTable.c
View File

0
source/dnode/vnode/meta/src/metaTbCfg.c → source/dnode/vnode/src/meta/metaTbCfg.c
View File

0
source/dnode/vnode/meta/src/metaTbTag.c → source/dnode/vnode/src/meta/metaTbTag.c
View File

0
source/dnode/vnode/meta/src/metaTbUid.c → source/dnode/vnode/src/meta/metaTbUid.c
View File

0
source/dnode/vnode/tq/src/tq.c → source/dnode/vnode/src/tq/tq.c
View File

0
source/dnode/vnode/tq/src/tqCommit.c → source/dnode/vnode/src/tq/tqCommit.c
View File

0
source/dnode/vnode/tq/src/tqMetaStore.c → source/dnode/vnode/src/tq/tqMetaStore.c
View File

0
source/dnode/vnode/tsdb/src/tsdbCommit.c → source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

0
source/dnode/vnode/tsdb/src/tsdbCompact.c → source/dnode/vnode/src/tsdb/tsdbCompact.c
View File

0
source/dnode/vnode/tsdb/src/tsdbFS.c → source/dnode/vnode/src/tsdb/tsdbFS.c
View File

0
source/dnode/vnode/tsdb/src/tsdbFile.c → source/dnode/vnode/src/tsdb/tsdbFile.c
View File

98
source/dnode/vnode/src/tsdb/tsdbHealth.c Normal file
View File

@ -0,0 +1,98 @@
/*
* 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/>.
*/
// #include "os.h"
// #include "tmsg.h"
// #include "tarray.h"
// #include "query.h"
// #include "tglobal.h"
// #include "tlist.h"
// #include "tsdbint.h"
// #include "tsdbBuffer.h"
// #include "tsdbLog.h"
// #include "tsdbHealth.h"
// #include "ttimer.h"
// #include "tthread.h"
// // return malloc new block count
// int32_t tsdbInsertNewBlock(STsdbRepo * pRepo) {
// STsdbBufPool *pPool = pRepo->pPool;
// int32_t cnt = 0;
// if(tsdbAllowNewBlock(pRepo)) {
// STsdbBufBlock *pBufBlock = tsdbNewBufBlock(pPool->bufBlockSize);
// if (pBufBlock) {
// if (tdListAppend(pPool->bufBlockList, (void *)(&pBufBlock)) < 0) {
// // append error
// tsdbFreeBufBlock(pBufBlock);
// } else {
// pPool->nElasticBlocks ++;
// cnt ++ ;
// }
// }
// }
// return cnt;
// }
// // switch anther thread to run
// void* cbKillQueryFree(void* param) {
// STsdbRepo* pRepo = (STsdbRepo*)param;
// // vnode
// if(pRepo->appH.notifyStatus) {
// pRepo->appH.notifyStatus(pRepo->appH.appH, TSDB_STATUS_COMMIT_NOBLOCK, TSDB_CODE_SUCCESS);
// }
// // free
// if(pRepo->pthread){
// void* p = pRepo->pthread;
// pRepo->pthread = NULL;
// free(p);
// }
// return NULL;
// }
// // return true do free , false do nothing
// bool tsdbUrgeQueryFree(STsdbRepo * pRepo) {
// // check previous running
// if(pRepo->pthread && taosThreadRunning(pRepo->pthread)) {
// tsdbWarn("vgId:%d pre urge thread is runing. nBlocks=%d nElasticBlocks=%d", REPO_ID(pRepo), pRepo->pPool->nBufBlocks, pRepo->pPool->nElasticBlocks);
// return false;
// }
// // create new
// pRepo->pthread = taosCreateThread(cbKillQueryFree, pRepo);
// if(pRepo->pthread == NULL) {
// tsdbError("vgId:%d create urge thread error.", REPO_ID(pRepo));
// return false;
// }
// return true;
// }
// bool tsdbAllowNewBlock(STsdbRepo* pRepo) {
// int32_t nMaxElastic = pRepo->config.totalBlocks/3;
// STsdbBufPool* pPool = pRepo->pPool;
// if(pPool->nElasticBlocks >= nMaxElastic) {
// tsdbWarn("vgId:%d tsdbAllowNewBlock return fasle. nElasticBlock(%d) >= MaxElasticBlocks(%d)", REPO_ID(pRepo), pPool->nElasticBlocks, nMaxElastic);
// return false;
// }
// return true;
// }
// bool tsdbNoProblem(STsdbRepo* pRepo) {
// if(listNEles(pRepo->pPool->bufBlockList) == 0)
// return false;
// return true;
// }

0
source/dnode/vnode/tsdb/src/tsdbMain.c → source/dnode/vnode/src/tsdb/tsdbMain.c
View File

Some files were not shown because too many files have changed in this diff Show More