Skip to content

TDengine
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
TDengine
TDengine
提交 78513e2e 编写于 作者: C Cary Xu
浏览文件
操作

Merge branch '3.0' into feature/TD-11463-3.0

上级 5443bb0d 74fb9433
展开全部 隐藏空白更改
内联 并排
Showing with 6881 addition and 3079 deletion
contrib/test/craft/CMakeLists.txt
浏览文件 @ 78513e2e
add_executable(simulate_vnode "simulate_vnode.c")
target_link_libraries(simulate_vnode craft lz4 uv_a)
\ No newline at end of file
target_link_libraries(simulate_vnode PUBLIC craft lz4 uv_a)
\ No newline at end of file
contrib/test/traft/single_node/CMakeLists.txt
浏览文件 @ 78513e2e
......@@ -3,4 +3,4 @@ target_sources(singleNode
PRIVATE
"singleNode.c"
)
target_link_libraries(singleNode traft lz4 uv_a)
target_link_libraries(singleNode PUBLIC traft lz4 uv_a)
include/common/tcommon.h
浏览文件 @ 78513e2e
......@@ -25,24 +25,6 @@
extern "C" {
#endif
// typedef struct STimeWindow {
// TSKEY skey;
// TSKEY ekey;
// } STimeWindow;
// typedef struct {
// int32_t dataLen;
// char name[TSDB_TABLE_FNAME_LEN];
// char *data;
// } STagData;
// typedef struct SSchema {
// uint8_t type;
// char name[TSDB_COL_NAME_LEN];
// int16_t colId;
// int16_t bytes;
// } SSchema;
enum {
TMQ_CONF__RESET_OFFSET__LATEST = -1,
TMQ_CONF__RESET_OFFSET__EARLIEAST = -2,
......@@ -50,7 +32,8 @@ enum {
};
enum {
TMQ_MSG_TYPE__POLL_RSP = 0,
TMQ_MSG_TYPE__DUMMY = 0,
TMQ_MSG_TYPE__POLL_RSP,
TMQ_MSG_TYPE__EP_RSP,
};
......@@ -285,4 +268,4 @@ typedef struct SSessionWindow {
}
#endif
#endif /*_TD_COMMON_DEF_H_*/
#endif /*_TD_COMMON_DEF_H_*/
include/common/tmsg.h
浏览文件 @ 78513e2e
......@@ -70,7 +70,7 @@ typedef uint16_t tmsg_t;
typedef enum {
HEARTBEAT_TYPE_MQ = 0,
HEARTBEAT_TYPE_QUERY = 1,
HEARTBEAT_TYPE_QUERY,
// types can be added here
//
HEARTBEAT_TYPE_MAX
......@@ -739,6 +739,9 @@ typedef struct {
int32_t maxRows;
int32_t commitTime;
int32_t fsyncPeriod;
uint32_t hashBegin;
uint32_t hashEnd;
int8_t hashMethod;
int8_t walLevel;
int8_t precision;
int8_t compression;
......@@ -749,6 +752,7 @@ typedef struct {
int8_t selfIndex;
int8_t streamMode;
SReplica replicas[TSDB_MAX_REPLICA];
} SCreateVnodeReq, SAlterVnodeReq;
int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
......
include/libs/executor/executor.h
浏览文件 @ 78513e2e
......@@ -89,25 +89,6 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t *useconds);
*/
int32_t qRetrieveQueryResultInfo(qTaskInfo_t tinfo, bool* buildRes, void* pRspContext);
/**
*
* Retrieve the actual results to fill the response message payload.
* Note that this function must be executed after qRetrieveQueryResultInfo is invoked.
*
* @param tinfo tinfo object
* @param pRsp response message
* @param contLen payload length
* @return
*/
//int32_t qDumpRetrieveResult(qTaskInfo_t tinfo, SRetrieveTableRsp** pRsp, int32_t* contLen, bool* continueExec);
/**
* return the transporter context (RPC)
* @param tinfo
* @return
*/
void* qGetResultRetrieveMsg(qTaskInfo_t tinfo);
/**
* kill the ongoing query and free the query handle and corresponding resources automatically
* @param tinfo qhandle
......
include/libs/index/index.h
浏览文件 @ 78513e2e
......@@ -29,6 +29,12 @@ typedef struct SIndexOpts SIndexOpts;
typedef struct SIndexMultiTermQuery SIndexMultiTermQuery;
typedef struct SArray SIndexMultiTerm;
typedef struct SIndex SIndexJson;
typedef struct SIndexTerm SIndexJsonTerm;
typedef struct SIndexOpts SIndexJsonOpts;
typedef struct SIndexMultiTermQuery SIndexJsonMultiTermQuery;
typedef struct SArray SIndexJsonMultiTerm;
typedef enum {
ADD_VALUE, // add index colume value
DEL_VALUE, // delete index column value
......@@ -39,24 +45,108 @@ typedef enum {
} SIndexOperOnColumn;
typedef enum { MUST = 0, SHOULD = 1, NOT = 2 } EIndexOperatorType;
typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2, QUERY_REGEX = 3 } EIndexQueryType;
typedef enum { QUERY_TERM = 0, QUERY_PREFIX = 1, QUERY_SUFFIX = 2, QUERY_REGEX = 3, QUERY_RANGE = 4 } EIndexQueryType;
/*
* @param: oper
*
* create multi query
* @param oper (input, relation between querys)
*/
SIndexMultiTermQuery* indexMultiTermQueryCreate(EIndexOperatorType oper);
void indexMultiTermQueryDestroy(SIndexMultiTermQuery* pQuery);
int indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EIndexQueryType type);
/*
* @param:
* @param:
* destroy multi query
* @param pQuery (input, multi-query-object to be destory)
*/
void indexMultiTermQueryDestroy(SIndexMultiTermQuery* pQuery);
/*
* add query to multi query
* @param pQuery (input, multi-query-object)
* @param term (input, single query term)
* @param type (input, single query type)
* @return error code
*/
int indexMultiTermQueryAdd(SIndexMultiTermQuery* pQuery, SIndexTerm* term, EIndexQueryType type);
/*
* open index
* @param opt (input, index opt)
* @param path (input, index path)
* @param index (output, index object)
* @return error code
*/
int indexOpen(SIndexOpts* opt, const char* path, SIndex** index);
/*
* close index
* @param index (input, index to be closed)
* @return error code
*/
int indexOpen(SIndexOpts* opt, const char* path, SIndex** index);
void indexClose(SIndex* index);
int indexPut(SIndex* index, SIndexMultiTerm* terms, uint64_t uid);
int indexDelete(SIndex* index, SIndexMultiTermQuery* query);
int indexSearch(SIndex* index, SIndexMultiTermQuery* query, SArray* result);
int indexRebuild(SIndex* index, SIndexOpts* opt);
/*
* insert terms into index
* @param index (input, index object)
* @param term (input, terms inserted into index)
* @param uid (input, uid of terms)
* @return error code
*/
int indexPut(SIndex* index, SIndexMultiTerm* terms, uint64_t uid);
/*
* delete terms that meet query condition
* @param index (input, index object)
* @param query (input, condition query to deleted)
* @return error code
*/
int indexDelete(SIndex* index, SIndexMultiTermQuery* query);
/*
* search index
* @param index (input, index object)
* @param query (input, multi query condition)
* @param result(output, query result)
* @return error code
*/
int indexSearch(SIndex* index, SIndexMultiTermQuery* query, SArray* result);
/*
* rebuild index
* @param index (input, index object)
* @parma opt (input, rebuild index opts)
* @return error code
*/
int indexRebuild(SIndex* index, SIndexOpts* opt);
/*
* open index
* @param opt (input,index json opt)
* @param path (input, index json path)
* @param index (output, index json object)
* @return error code
*/
int tIndexJsonOpen(SIndexJsonOpts* opts, const char* path, SIndexJson** index);
/*
* close index
* @param index (input, index to be closed)
* @return void
*/
void tIndexJsonClose(SIndexJson* index);
/*
* insert terms into index
* @param index (input, index object)
* @param term (input, terms inserted into index)
* @param uid (input, uid of terms)
* @return error code
*/
int tIndexJsonPut(SIndexJson* index, SIndexJsonMultiTerm* terms, uint64_t uid);
/*
* search index
* @param index (input, index object)
* @param query (input, multi query condition)
* @param result(output, query result)
* @return error code
*/
int tIndexJsonSearch(SIndexJson* index, SIndexJsonMultiTermQuery* query, SArray* result);
/*
* @param
* @param
......
include/libs/monitor/monitor.h
浏览文件 @ 78513e2e
......@@ -18,11 +18,17 @@
#include "tarray.h"
#include "tdef.h"
#include "tlog.h"
#ifdef __cplusplus
extern "C" {
#endif
#define MON_STATUS_LEN 8
#define MON_ROLE_LEN 9
#define MON_VER_LEN 12
#define MON_LOG_LEN 1024
typedef struct {
int32_t dnode_id;
char dnode_ep[TSDB_EP_LEN];
......@@ -31,19 +37,19 @@ typedef struct {
typedef struct {
int32_t dnode_id;
char dnode_ep[TSDB_EP_LEN];
char status[8];
char status[MON_STATUS_LEN];
} SMonDnodeDesc;
typedef struct {
int32_t mnode_id;
char mnode_ep[TSDB_EP_LEN];
char role[8];
char role[MON_ROLE_LEN];
} SMonMnodeDesc;
typedef struct {
char first_ep[TSDB_EP_LEN];
int32_t first_ep_dnode_id;
char version[12];
char version[MON_VER_LEN];
float master_uptime; // day
int32_t monitor_interval; // sec
int32_t vgroups_total;
......@@ -57,19 +63,18 @@ typedef struct {
typedef struct {
int32_t dnode_id;
int8_t vnode_online;
char vnode_role[8];
char vnode_role[MON_ROLE_LEN];
} SMonVnodeDesc;
typedef struct {
int32_t vgroup_id;
char database_name[TSDB_DB_NAME_LEN];
int32_t tables_num;
char status[MON_STATUS_LEN];
SMonVnodeDesc vnodes[TSDB_MAX_REPLICA];
} SMonVgroupDesc;
typedef struct {
char database_name[TSDB_DB_NAME_LEN];
int32_t tables_num;
int8_t status;
SArray *vgroups; // array of SMonVgroupDesc
} SMonVgroupInfo;
......@@ -84,18 +89,18 @@ typedef struct {
float cpu_engine;
float cpu_system;
float cpu_cores;
float mem_engine; // MB
float mem_system; // MB
float mem_total; // MB
int64_t mem_engine; // KB
int64_t mem_system; // KB
int64_t mem_total; // KB
float disk_engine; // GB
float disk_used; // GB
float disk_total; // GB
float net_in; // Kb/s
float net_out; // Kb/s
float io_read; // Mb/s
float io_write; // Mb/s
float io_read_disk; // Mb/s
float io_write_disk; // Mb/s
int64_t net_in;
int64_t net_out;
float io_read;
float io_write;
float io_read_disk;
float io_write_disk;
int32_t req_select;
float req_select_rate;
int32_t req_insert;
......@@ -107,25 +112,21 @@ typedef struct {
int32_t errors;
int32_t vnodes_num;
int32_t masters;
int32_t has_mnode;
int8_t has_mnode;
} SMonDnodeInfo;
typedef struct {
char name[TSDB_FILENAME_LEN];
int32_t level;
int8_t level;
SDiskSize size;
} SMonDiskDesc;
typedef struct {
SArray *disks; // array of SMonDiskDesc
SArray *datadirs; // array of SMonDiskDesc
SMonDiskDesc logdir;
SMonDiskDesc tempdir;
} SMonDiskInfo;
typedef struct {
int64_t ts;
int8_t level;
char content[1024];
} SMonLogItem;
typedef struct SMonInfo SMonInfo;
typedef struct {
......@@ -136,7 +137,7 @@ typedef struct {
int32_t monInit(const SMonCfg *pCfg);
void monCleanup();
void monAddLogItem(SMonLogItem *pItem);
void monRecordLog(int64_t ts, ELogLevel level, const char *content);
SMonInfo *monCreateMonitorInfo();
void monSetBasicInfo(SMonInfo *pMonitor, SMonBasicInfo *pInfo);
......
include/libs/qcom/query.h
浏览文件 @ 78513e2e
......@@ -168,20 +168,18 @@ int32_t queryCreateTableMetaFromMsg(STableMetaRsp* msg, bool isSuperTable, STabl
extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char **msg, int32_t msgSize, int32_t *msgLen);
extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char *msg, int32_t msgSize);
#define SET_META_TYPE_NULL(t) (t) = META_TYPE_NULL_TABLE
#define SET_META_TYPE_CTABLE(t) (t) = META_TYPE_CTABLE
#define SET_META_TYPE_TABLE(t) (t) = META_TYPE_TABLE
#define SET_META_TYPE_BOTH_TABLE(t) (t) = META_TYPE_BOTH_TABLE
#define qFatal(...) do { if (qDebugFlag & DEBUG_FATAL) { taosPrintLog("QRY FATAL ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qError(...) do { if (qDebugFlag & DEBUG_ERROR) { taosPrintLog("QRY ERROR ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qWarn(...) do { if (qDebugFlag & DEBUG_WARN) { taosPrintLog("QRY WARN ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qInfo(...) do { if (qDebugFlag & DEBUG_INFO) { taosPrintLog("QRY ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qDebug(...) do { if (qDebugFlag & DEBUG_DEBUG) { taosPrintLog("QRY ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qTrace(...) do { if (qDebugFlag & DEBUG_TRACE) { taosPrintLog("QRY ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qDebugL(...) do { if (qDebugFlag & DEBUG_DEBUG) { taosPrintLongString("QRY ", qDebugFlag, __VA_ARGS__); }} while(0)
#define qFatal(...) do { if (qDebugFlag & DEBUG_FATAL) { taosPrintLog("QRY FATAL ", DEBUG_FATAL, qDebugFlag, __VA_ARGS__); }} while(0)
#define qError(...) do { if (qDebugFlag & DEBUG_ERROR) { taosPrintLog("QRY ERROR ", DEBUG_ERROR, qDebugFlag, __VA_ARGS__); }} while(0)
#define qWarn(...) do { if (qDebugFlag & DEBUG_WARN) { taosPrintLog("QRY WARN ", DEBUG_WARN, qDebugFlag, __VA_ARGS__); }} while(0)
#define qInfo(...) do { if (qDebugFlag & DEBUG_INFO) { taosPrintLog("QRY ", DEBUG_INFO, qDebugFlag, __VA_ARGS__); }} while(0)
#define qDebug(...) do { if (qDebugFlag & DEBUG_DEBUG) { taosPrintLog("QRY ", DEBUG_DEBUG, qDebugFlag, __VA_ARGS__); }} while(0)
#define qTrace(...) do { if (qDebugFlag & DEBUG_TRACE) { taosPrintLog("QRY ", DEBUG_TRACE, qDebugFlag, __VA_ARGS__); }} while(0)
#define qDebugL(...) do { if (qDebugFlag & DEBUG_DEBUG) { taosPrintLongString("QRY ", DEBUG_DEBUG, qDebugFlag, __VA_ARGS__); }} while(0)
#ifdef __cplusplus
}
......
include/libs/scalar/filter.h
浏览文件 @ 78513e2e
......@@ -19,10 +19,12 @@
extern "C" {
#endif
#include "tcommon.h"
#include "nodes.h"
typedef struct SFilterInfo SFilterInfo;
typedef int32_t (*filer_get_col_from_id)(void *, int32_t, void **);
enum {
FLT_OPTION_NO_REWRITE = 1,
FLT_OPTION_TIMESTAMP = 2,
......@@ -34,7 +36,6 @@ typedef struct SFilterColumnParam{
SArray* pDataBlock;
} SFilterColumnParam;
extern int32_t filterInitFromNode(SNode *pNode, SFilterInfo **pinfo, uint32_t options);
extern bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols);
extern int32_t filterSetDataFromSlotId(SFilterInfo *info, void *param);
......
include/libs/scheduler/scheduler.h
浏览文件 @ 78513e2e
......@@ -23,8 +23,6 @@ extern "C" {
#include "catalog.h"
#include "planner.h"
struct SSchJob;
typedef struct SSchedulerCfg {
uint32_t maxJobNum;
} SSchedulerCfg;
......@@ -72,7 +70,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg);
* @param nodeList Qnode/Vnode address list, element is SQueryNodeAddr
* @return
*/
int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, struct SSchJob** pJob, const char* sql, SQueryResult *pRes);
int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, int64_t *pJob, const char* sql, SQueryResult *pRes);
/**
* Process the query job, generated according to the query physical plan.
......@@ -80,7 +78,7 @@ int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryDag* pDag, str
* @param pNodeList Qnode/Vnode address list, element is SQueryNodeAddr
* @return
*/
int32_t schedulerAsyncExecJob(void *transport, SArray *pNodeList, SQueryDag* pDag, const char* sql, struct SSchJob** pJob);
int32_t schedulerAsyncExecJob(void *transport, SArray *pNodeList, SQueryDag* pDag, const char* sql, int64_t *pJob);
/**
* Fetch query result from the remote query executor
......@@ -88,7 +86,7 @@ int32_t schedulerAsyncExecJob(void *transport, SArray *pNodeList, SQueryDag* pDa
* @param data
* @return
*/
int32_t schedulerFetchRows(struct SSchJob *pJob, void **data);
int32_t schedulerFetchRows(int64_t job, void **data);
/**
......@@ -102,7 +100,7 @@ int32_t schedulerFetchRows(struct SSchJob *pJob, void **data);
* Free the query job
* @param pJob
*/
void schedulerFreeJob(void *pJob);
void schedulerFreeJob(int64_t job);
void schedulerDestroy(void);
......
include/libs/sync/sync.h
浏览文件 @ 78513e2e
......@@ -21,7 +21,7 @@ extern "C" {
#endif
#include <stdint.h>
#include <tep.h>
#include <tdatablock.h>
#include "taosdef.h"
#include "trpc.h"
......@@ -34,9 +34,7 @@ typedef enum {
TAOS_SYNC_STATE_FOLLOWER = 0,
TAOS_SYNC_STATE_CANDIDATE = 1,
TAOS_SYNC_STATE_LEADER = 2,
} ESyncRole;
typedef ESyncRole ESyncState;
} ESyncState;
typedef struct SSyncBuffer {
void* data;
......
include/libs/tfs/tfs.h
浏览文件 @ 78513e2e
......@@ -17,6 +17,7 @@
#define _TD_TFS_H_
#include "tdef.h"
#include "monitor.h"
#ifdef __cplusplus
extern "C" {
......@@ -237,6 +238,14 @@ const STfsFile *tfsReaddir(STfsDir *pDir);
*/
void tfsClosedir(STfsDir *pDir);
/**
* @brief Get disk info of tfs.
*
* @param pTfs The fs object.
* @param pInfo The info object.
*/
int32_t tfsGetMonitorInfo(STfs *pTfs, SMonDiskInfo *pInfo);
#ifdef __cplusplus
}
#endif
......
include/libs/wal/wal.h
浏览文件 @ 78513e2e
......@@ -24,43 +24,41 @@
extern "C" {
#endif
extern int32_t wDebugFlag;
#define wFatal(...) \
{ \
if (wDebugFlag & DEBUG_FATAL) { \
taosPrintLog("WAL FATAL ", 255, __VA_ARGS__); \
} \
#define wFatal(...) \
{ \
if (wDebugFlag & DEBUG_FATAL) { \
taosPrintLog("WAL FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \
} \
}
#define wError(...) \
{ \
if (wDebugFlag & DEBUG_ERROR) { \
taosPrintLog("WAL ERROR ", 255, __VA_ARGS__); \
} \
#define wError(...) \
{ \
if (wDebugFlag & DEBUG_ERROR) { \
taosPrintLog("WAL ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \
} \
}
#define wWarn(...) \
{ \
if (wDebugFlag & DEBUG_WARN) { \
taosPrintLog("WAL WARN ", 255, __VA_ARGS__); \
} \
#define wWarn(...) \
{ \
if (wDebugFlag & DEBUG_WARN) { \
taosPrintLog("WAL WARN ", DEBUG_WARN, 255, __VA_ARGS__); \
} \
}
#define wInfo(...) \
{ \
if (wDebugFlag & DEBUG_INFO) { \
taosPrintLog("WAL ", 255, __VA_ARGS__); \
} \
#define wInfo(...) \
{ \
if (wDebugFlag & DEBUG_INFO) { \
taosPrintLog("WAL ", DEBUG_INFO, 255, __VA_ARGS__); \
} \
}
#define wDebug(...) \
{ \
if (wDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); \
} \
#define wDebug(...) \
{ \
if (wDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("WAL ", DEBUG_DEBUG, wDebugFlag, __VA_ARGS__); \
} \
}
#define wTrace(...) \
{ \
if (wDebugFlag & DEBUG_TRACE) { \
taosPrintLog("WAL ", wDebugFlag, __VA_ARGS__); \
} \
#define wTrace(...) \
{ \
if (wDebugFlag & DEBUG_TRACE) { \
taosPrintLog("WAL ", DEBUG_TRACE, wDebugFlag, __VA_ARGS__); \
} \
}
#define WAL_HEAD_VER 0
......
include/os/os.h
浏览文件 @ 78513e2e
......@@ -44,6 +44,7 @@ extern "C" {
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/utsname.h>
#include <sys/param.h>
#include <unistd.h>
#include <wchar.h>
#include <wctype.h>
......
include/os/osDir.h
浏览文件 @ 78513e2e
......@@ -21,7 +21,7 @@ extern "C" {
#endif
void taosRemoveDir(const char *dirname);
int32_t taosDirExist(char *dirname);
bool taosDirExist(char *dirname);
int32_t taosMkDir(const char *dirname);
void taosRemoveOldFiles(const char *dirname, int32_t keepDays);
int32_t taosExpandDir(const char *dirname, char *outname, int32_t maxlen);
......
include/os/osEnv.h
浏览文件 @ 78513e2e
......@@ -28,11 +28,11 @@ extern char tsCharset[];
extern char tsLocale[];
extern int8_t tsDaylight;
extern bool tsEnableCoreFile;
extern int64_t tsPageSize;
extern int64_t tsPageSizeKB;
extern int64_t tsOpenMax;
extern int64_t tsStreamMax;
extern int32_t tsNumOfCores;
extern int32_t tsTotalMemoryMB;
extern float tsNumOfCores;
extern int64_t tsTotalMemoryKB;
extern char configDir[];
extern char tsDataDir[];
......
include/os/osFile.h
浏览文件 @ 78513e2e
......@@ -22,11 +22,18 @@ extern "C" {
#include "osSocket.h"
// If the error is in a third-party library, place this header file under the third-party library header file.
#ifndef ALLOW_FORBID_FUNC
#define open OPEN_FUNC_TAOS_FORBID
#define fopen FOPEN_FUNC_TAOS_FORBID
// #define close CLOSE_FUNC_TAOS_FORBID
// #define fclose FCLOSE_FUNC_TAOS_FORBID
#define access ACCESS_FUNC_TAOS_FORBID
#define stat STAT_FUNC_TAOS_FORBID
#define lstat LSTAT_FUNC_TAOS_FORBID
#define fstat FSTAT_FUNC_TAOS_FORBID
#define close CLOSE_FUNC_TAOS_FORBID
#define fclose FCLOSE_FUNC_TAOS_FORBID
#define fsync FSYNC_FUNC_TAOS_FORBID
// #define fflush FFLUSH_FUNC_TAOS_FORBID
#endif
#ifndef PATH_MAX
......@@ -43,7 +50,13 @@ typedef struct TdFile *TdFilePtr;
#define TD_FILE_TEXT 0x0020
#define TD_FILE_AUTO_DEL 0x0040
#define TD_FILE_EXCL 0x0080
#define TD_FILE_STREAM 0x0100 // Only support taosFprintfFile, taosGetLineFile, taosGetLineFile, taosEOFFile
#define TD_FILE_STREAM 0x0100 // Only support taosFprintfFile, taosGetLineFile, taosEOFFile
TdFilePtr taosOpenFile(const char *path,int32_t tdFileOptions);
#define TD_FILE_ACCESS_EXIST_OK 0x1
#define TD_FILE_ACCESS_READ_OK 0x2
#define TD_FILE_ACCESS_WRITE_OK 0x4
bool taosCheckAccessFile(const char *pathname, int mode);
int32_t taosLockFile(TdFilePtr pFile);
int32_t taosUnLockFile(TdFilePtr pFile);
......@@ -51,9 +64,9 @@ int32_t taosUnLockFile(TdFilePtr pFile);
int32_t taosUmaskFile(int32_t maskVal);
int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime);
int32_t taosDevInoFile(const char *path, int64_t *stDev, int64_t *stIno);
int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime);
TdFilePtr taosOpenFile(const char *path,int32_t tdFileOptions);
bool taosCheckExistFile(const char *pathname);
int64_t taosLSeekFile(TdFilePtr pFile, int64_t offset, int32_t whence);
int32_t taosFtruncateFile(TdFilePtr pFile, int64_t length);
......@@ -62,7 +75,7 @@ int32_t taosFsyncFile(TdFilePtr pFile);
int64_t taosReadFile(TdFilePtr pFile, void *buf, int64_t count);
int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset);
int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count);
void taosFprintfFile(TdFilePtr pFile, const char *format, ...);
void taosFprintfFile(TdFilePtr pFile, const char *format, ...);
int64_t taosGetLineFile(TdFilePtr pFile, char ** __restrict__ ptrBuf);
int32_t taosEOFFile(TdFilePtr pFile);
......@@ -70,8 +83,9 @@ int64_t taosCloseFile(TdFilePtr *ppFile);
int32_t taosRenameFile(const char *oldName, const char *newName);
int64_t taosCopyFile(const char *from, const char *to);
int32_t taosRemoveFile(const char *path);
void taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath);
void taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath);
int64_t taosSendFile(SocketFd fdDst, TdFilePtr pFileSrc, int64_t *offset, int64_t size);
int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size);
......@@ -79,7 +93,7 @@ int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, in
void *taosMmapReadOnlyFile(TdFilePtr pFile, int64_t length);
bool taosValidFile(TdFilePtr pFile);
int taosGetErrorFile(TdFilePtr pFile);
int32_t taosGetErrorFile(TdFilePtr pFile);
#ifdef __cplusplus
}
......
include/os/osSysinfo.h
浏览文件 @ 78513e2e
......@@ -34,19 +34,19 @@ typedef struct {
} SDiskSpace;
void taosGetSystemInfo();
bool taosGetEmail(char *email, int32_t maxLen);
bool taosGetOsReleaseName(char *releaseName, int32_t maxLen);
bool taosGetCpuInfo(char *cpuModel, int32_t maxLen, int32_t *numOfCores);
int32_t taosGetCpuCores();
bool taosGetCpuUsage(float *sysCpuUsage, float *procCpuUsage);
bool taosGetTotalSysMemoryKB(uint64_t *kb);
bool taosGetProcMemory(float *memoryUsedMB); //
bool taosGetSysMemory(float *memoryUsedMB); //
int32_t taosGetEmail(char *email, int32_t maxLen);
int32_t taosGetOsReleaseName(char *releaseName, int32_t maxLen);
int32_t taosGetCpuInfo(char *cpuModel, int32_t maxLen, float *numOfCores);
int32_t taosGetCpuCores(float *numOfCores);
int32_t taosGetCpuUsage(float *cpu_system, float *cpu_engine);
int32_t taosGetTotalMemory(int64_t *totalKB);
int32_t taosGetProcMemory(int64_t *usedKB);
int32_t taosGetSysMemory(int64_t *usedKB);
int32_t taosGetDiskSize(char *dataDir, SDiskSize *diskSize);
bool taosReadProcIO(int64_t *rchars, int64_t *wchars);
bool taosGetProcIO(float *readKB, float *writeKB);
bool taosGetCardInfo(int64_t *bytes, int64_t *rbytes, int64_t *tbytes);
bool taosGetBandSpeed(float *bandSpeedKb);
int32_t taosReadProcIO(int64_t *rchars, int64_t *wchars);
int32_t taosGetProcIO(float *readKB, float *writeKB);
int32_t taosGetCardInfo(int64_t *bytes, int64_t *rbytes, int64_t *tbytes);
int32_t taosGetBandSpeed(float *bandSpeedKb);
int32_t taosSystem(const char *cmd);
void taosKillSystem();
......
include/util/tcache.h
浏览文件 @ 78513e2e
......@@ -40,55 +40,9 @@ typedef struct SCacheStatis {
int64_t refreshCount;
} SCacheStatis;
struct STrashElem;
typedef struct SCacheDataNode {
uint64_t addedTime; // the added time when this element is added or updated into cache
uint64_t lifespan; // life duration when this element should be remove from cache
uint64_t expireTime; // expire time
uint64_t signature;
struct STrashElem *pTNodeHeader; // point to trash node head
uint16_t keySize : 15; // max key size: 32kb
bool inTrashcan : 1; // denote if it is in trash or not
uint32_t size; // allocated size for current SCacheDataNode
T_REF_DECLARE()
char *key;
char data[];
} SCacheDataNode;
typedef struct STrashElem {
struct STrashElem *prev;
struct STrashElem *next;
SCacheDataNode *pData;
} STrashElem;
/*
* to accommodate the old data which has the same key value of new one in hashList
* when an new node is put into cache, if an existed one with the same key:
* 1. if the old one does not be referenced, update it.
* 2. otherwise, move the old one to pTrash, addedTime the new one.
*
* when the node in pTrash does not be referenced, it will be release at the expired expiredTime
*/
typedef struct {
int64_t totalSize; // total allocated buffer in this hash table, SCacheObj is not included.
int64_t refreshTime;
STrashElem *pTrash;
char *name;
SCacheStatis statistics;
SHashObj *pHashTable;
__cache_free_fn_t freeFp;
uint32_t numOfElemsInTrash; // number of element in trash
uint8_t deleting; // set the deleting flag to stop refreshing ASAP.
pthread_t refreshWorker;
bool extendLifespan; // auto extend life span when one item is accessed.
int64_t checkTick; // tick used to record the check times of the refresh threads
#if defined(LINUX)
pthread_rwlock_t lock;
#else
pthread_mutex_t lock;
#endif
} SCacheObj;
typedef struct SCacheObj SCacheObj;
typedef struct SCacheIter SCacheIter;
typedef struct STrashElem STrashElem;
/**
* initialize the cache object
......@@ -141,7 +95,7 @@ void *taosCacheAcquireByData(SCacheObj *pCacheObj, void *data);
* @param data
* @return
*/
void *taosCacheTransfer(SCacheObj *pCacheObj, void **data);
void *taosCacheTransferData(SCacheObj *pCacheObj, void **data);
/**
* remove data in cache, the data will not be removed immediately.
......@@ -152,6 +106,13 @@ void *taosCacheTransfer(SCacheObj *pCacheObj, void **data);
*/
void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove);
/**
*
* @param pCacheObj
* @return
*/
size_t taosCacheGetNumOfObj(const SCacheObj* pCacheObj);
/**
* move all data node into trash, clear node in trash can if it is not referenced by any clients
* @param handle
......@@ -184,6 +145,12 @@ void taosCacheRefresh(SCacheObj *pCacheObj, __cache_trav_fn_t fp, void *param1);
*/
void taosStopCacheRefreshWorker();
SCacheIter* taosCacheCreateIter(const SCacheObj* pCacheObj);
bool taosCacheIterNext(SCacheIter* pIter);
void* taosCacheIterGetData(const SCacheIter* pIter, size_t* dataLen);
void* taosCacheIterGetKey(const SCacheIter* pIter, size_t* keyLen);
void taosCacheDestroyIter(SCacheIter* pIter);
#ifdef __cplusplus
}
#endif
......
include/util/thash.h
浏览文件 @ 78513e2e
......@@ -28,12 +28,8 @@ typedef int32_t (*_equal_fn_t)(const void *, const void *, size_t len);
typedef void (*_hash_before_fn_t)(void *);
typedef void (*_hash_free_fn_t)(void *);
#define HASH_MAX_CAPACITY (1024 * 1024 * 16)
#define HASH_DEFAULT_LOAD_FACTOR (0.75)
#define HASH_INDEX(v, c) ((v) & ((c)-1))
#define HASH_NODE_EXIST(code) (code == -2)
#define HASH_KEY_ALREADY_EXISTS (-2)
#define HASH_NODE_EXIST(code) (code == HASH_KEY_ALREADY_EXISTS)
/**
* murmur hash algorithm
......@@ -54,49 +50,15 @@ uint32_t taosIntHash_32(const char *key, uint32_t len);
uint32_t taosIntHash_64(const char *key, uint32_t len);
_hash_fn_t taosGetDefaultHashFunction(int32_t type);
_equal_fn_t taosGetDefaultEqualFunction(int32_t type);
typedef struct SHashNode {
struct SHashNode *next;
uint32_t hashVal; // the hash value of key
uint32_t dataLen; // length of data
uint32_t keyLen; // length of the key
uint16_t count; // reference count
int8_t removed; // flag to indicate removed
char data[];
} SHashNode;
#define GET_HASH_NODE_KEY(_n) ((char *)(_n) + sizeof(SHashNode) + (_n)->dataLen)
#define GET_HASH_NODE_DATA(_n) ((char *)(_n) + sizeof(SHashNode))
#define GET_HASH_PNODE(_n) ((SHashNode *)((char *)(_n) - sizeof(SHashNode)))
typedef enum SHashLockTypeE {
HASH_NO_LOCK = 0,
HASH_ENTRY_LOCK = 1,
} SHashLockTypeE;
typedef struct SHashEntry {
int32_t num; // number of elements in current entry
SRWLatch latch; // entry latch
SHashNode *next;
} SHashEntry;
typedef struct SHashObj {
SHashEntry **hashList;
uint32_t capacity; // number of slots
uint32_t size; // number of elements in hash table
_hash_fn_t hashFp; // hash function
_hash_free_fn_t freeFp; // hash node free callback function
_equal_fn_t equalFp; // equal function
_hash_before_fn_t callbackFp; // function invoked before return the value to caller
SRWLatch lock; // read-write spin lock
SHashLockTypeE type; // lock type
bool enableUpdate; // enable update
SArray *pMemBlock; // memory block allocated for SHashEntry
} SHashObj;
typedef struct SHashNode SHashNode;
typedef struct SHashObj SHashObj;
/**
* init the hash table
......@@ -126,8 +88,6 @@ int32_t taosHashGetSize(const SHashObj *pHashObj);
*/
int32_t taosHashPut(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size);
int32_t taosHashPutExt(SHashObj *pHashObj, const void *key, size_t keyLen, void *data, size_t size, bool *newAdded);
/**
* return the payload data with the specified key
*
......@@ -146,17 +106,18 @@ void *taosHashGet(SHashObj *pHashObj, const void *key, size_t keyLen);
* @param destBuf
* @return
*/
void *taosHashGetClone(SHashObj *pHashObj, const void *key, size_t keyLen, void *destBuf);
int32_t taosHashGetDup(SHashObj *pHashObj, const void *key, size_t keyLen, void *destBuf);
/**
* Clone the result to interval allocated buffer
*
* @param pHashObj
* @param key
* @param keyLen
* @param destBuf
* @param size
* @return
*/
void *taosHashGetCloneExt(SHashObj *pHashObj, const void *key, size_t keyLen, void (*fp)(void *), void **d, size_t *sz);
int32_t taosHashGetDup_m(SHashObj* pHashObj, const void* key, size_t keyLen, void** destBuf, int32_t* size);
/**
* remove item with the specified key
......@@ -207,37 +168,13 @@ void *taosHashIterate(SHashObj *pHashObj, void *p);
*/
void taosHashCancelIterate(SHashObj *pHashObj, void *p);
/**
* Get the corresponding key information for a given data in hash table
* @param data
* @return
*/
int32_t taosHashGetKey(void *data, void **key, size_t *keyLen);
/**
* Get the corresponding key information for a given data in hash table, using memcpy
* @param data
* @param dst
* @return
*/
static FORCE_INLINE int32_t taosHashCopyKey(void *data, void *dst) {
if (NULL == data || NULL == dst) {
return -1;
}
SHashNode *node = GET_HASH_PNODE(data);
void *key = GET_HASH_NODE_KEY(node);
memcpy(dst, key, node->keyLen);
return 0;
}
/**
* Get the corresponding data length for a given data in hash table
* @param data
* @return
*/
int32_t taosHashGetDataLen(void *data);
/**
* Get the corresponding key information for a given data in hash table
* @param data
* @param keyLen
* @return
*/
void *taosHashGetKey(void *data, size_t* keyLen);
/**
* return the payload data with the specified key(reference number added)
......@@ -258,8 +195,20 @@ void *taosHashAcquire(SHashObj *pHashObj, const void *key, size_t keyLen);
*/
void taosHashRelease(SHashObj *pHashObj, void *p);
/**
*
* @param pHashObj
* @param fp
*/
void taosHashSetEqualFp(SHashObj *pHashObj, _equal_fn_t fp);
/**
*
* @param pHashObj
* @param fp
*/
void taosHashSetFreeFp(SHashObj *pHashObj, _hash_free_fn_t fp);
#ifdef __cplusplus
}
#endif
......
include/util/tlog.h
浏览文件 @ 78513e2e
......@@ -22,9 +22,29 @@
extern "C" {
#endif
typedef enum {
DEBUG_FATAL = 1,
DEBUG_ERROR = 1,
DEBUG_WARN = 2,
DEBUG_INFO = 2,
DEBUG_DEBUG = 4,
DEBUG_TRACE = 8,
DEBUG_DUMP = 16,
DEBUG_SCREEN = 64,
DEBUG_FILE = 128
} ELogLevel;
typedef void (*LogFp)(int64_t ts, ELogLevel level, const char *content);
extern bool tsLogEmbedded;
extern bool tsAsyncLog;
extern int32_t tsNumOfLogLines;
extern int32_t tsLogKeepDays;
extern LogFp tsLogFp;
extern int64_t tsNumOfErrorLogs;
extern int64_t tsNumOfInfoLogs;
extern int64_t tsNumOfDebugLogs;
extern int64_t tsNumOfTraceLogs;
extern int32_t dDebugFlag;
extern int32_t vDebugFlag;
extern int32_t mDebugFlag;
......@@ -40,45 +60,33 @@ extern int32_t tsdbDebugFlag;
extern int32_t tqDebugFlag;
extern int32_t fsDebugFlag;
#define DEBUG_FATAL 1U
#define DEBUG_ERROR DEBUG_FATAL
#define DEBUG_WARN 2U
#define DEBUG_INFO DEBUG_WARN
#define DEBUG_DEBUG 4U
#define DEBUG_TRACE 8U
#define DEBUG_DUMP 16U
#define DEBUG_SCREEN 64U
#define DEBUG_FILE 128U
int32_t taosInitLog(const char *logName, int32_t maxFiles);
void taosCloseLog();
void taosResetLog();
void taosSetAllDebugFlag(int32_t flag);
void taosDumpData(uint8_t *msg, int32_t len);
void taosPrintLog(const char *flags, int32_t dflag, const char *format, ...)
void taosPrintLog(const char *flags, ELogLevel level, int32_t dflag, const char *format, ...)
#ifdef __GNUC__
__attribute__((format(printf, 3, 4)))
__attribute__((format(printf, 4, 5)))
#endif
;
void taosPrintLongString(const char *flags, int32_t dflag, const char *format, ...)
void taosPrintLongString(const char *flags, ELogLevel level, int32_t dflag, const char *format, ...)
#ifdef __GNUC__
__attribute__((format(printf, 3, 4)))
__attribute__((format(printf, 4, 5)))
#endif
;
extern int8_t tscEmbeddedInUtil;
#define uFatal(...) { if (uDebugFlag & DEBUG_FATAL) { taosPrintLog("UTL FATAL", tscEmbeddedInUtil ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uError(...) { if (uDebugFlag & DEBUG_ERROR) { taosPrintLog("UTL ERROR ", tscEmbeddedInUtil ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uWarn(...) { if (uDebugFlag & DEBUG_WARN) { taosPrintLog("UTL WARN ", tscEmbeddedInUtil ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uInfo(...) { if (uDebugFlag & DEBUG_INFO) { taosPrintLog("UTL ", tscEmbeddedInUtil ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uDebug(...) { if (uDebugFlag & DEBUG_DEBUG) { taosPrintLog("UTL ", uDebugFlag, __VA_ARGS__); }}
#define uTrace(...) { if (uDebugFlag & DEBUG_TRACE) { taosPrintLog("UTL ", uDebugFlag, __VA_ARGS__); }}
#define uFatal(...) { if (uDebugFlag & DEBUG_FATAL) { taosPrintLog("UTL FATAL", DEBUG_FATAL, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uError(...) { if (uDebugFlag & DEBUG_ERROR) { taosPrintLog("UTL ERROR ", DEBUG_ERROR, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uWarn(...) { if (uDebugFlag & DEBUG_WARN) { taosPrintLog("UTL WARN ", DEBUG_WARN, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uInfo(...) { if (uDebugFlag & DEBUG_INFO) { taosPrintLog("UTL ", DEBUG_INFO, tsLogEmbedded ? 255 : uDebugFlag, __VA_ARGS__); }}
#define uDebug(...) { if (uDebugFlag & DEBUG_DEBUG) { taosPrintLog("UTL ", DEBUG_DEBUG, uDebugFlag, __VA_ARGS__); }}
#define uTrace(...) { if (uDebugFlag & DEBUG_TRACE) { taosPrintLog("UTL ", DEBUG_TRACE, uDebugFlag, __VA_ARGS__); }}
#define pError(...) { taosPrintLog("APP ERROR ", 255, __VA_ARGS__); }
#define pPrint(...) { taosPrintLog("APP ", 255, __VA_ARGS__); }
#define pError(...) { taosPrintLog("APP ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }
#define pPrint(...) { taosPrintLog("APP ", DEBUG_INFO, 255, __VA_ARGS__); }
#ifdef __cplusplus
}
......
include/util/tpagedbuf.h
浏览文件 @ 78513e2e
......@@ -53,7 +53,7 @@ typedef struct SDiskbasedBufStatis {
* @param handle
* @return
*/
int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMemBufSize, uint64_t qId, const char* dir);
int32_t createDiskbasedBuf(SDiskbasedBuf** pBuf, int32_t pagesize, int32_t inMemBufSize, const char* id, const char* dir);
/**
*
......
source/client/inc/clientInt.h
浏览文件 @ 78513e2e
......@@ -20,12 +20,12 @@
extern "C" {
#endif
#include "tcommon.h"
#include "parser.h"
#include "query.h"
#include "taos.h"
#include "tcommon.h"
#include "tdatablock.h"
#include "tdef.h"
#include "tep.h"
#include "thash.h"
#include "tlist.h"
#include "tmsg.h"
......@@ -47,12 +47,12 @@ extern "C" {
typedef struct SAppInstInfo SAppInstInfo;
typedef struct SHbConnInfo {
typedef struct {
void* param;
SClientHbReq* req;
} SHbConnInfo;
typedef struct SAppHbMgr {
typedef struct {
char* key;
// statistics
int32_t reportCnt;
......@@ -68,11 +68,11 @@ typedef struct SAppHbMgr {
SHashObj* connInfo; // hash<SClientHbKey, SHbConnInfo>
} SAppHbMgr;
typedef int32_t (*FHbRspHandle)(struct SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp);
typedef int32_t (*FHbRspHandle)(SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp);
typedef int32_t (*FHbReqHandle)(SClientHbKey* connKey, void* param, SClientHbReq* req);
typedef struct SClientHbMgr {
typedef struct {
int8_t inited;
// ctl
int8_t threadStop;
......@@ -108,13 +108,13 @@ typedef struct SHeartBeatInfo {
} SHeartBeatInfo;
struct SAppInstInfo {
int64_t numOfConns;
SCorEpSet mgmtEp;
SInstanceSummary summary;
SList* pConnList; // STscObj linked list
int64_t clusterId;
void* pTransporter;
struct SAppHbMgr* pAppHbMgr;
int64_t numOfConns;
SCorEpSet mgmtEp;
SInstanceSummary summary;
SList* pConnList; // STscObj linked list
int64_t clusterId;
void* pTransporter;
SAppHbMgr* pAppHbMgr;
};
typedef struct SAppInfo {
......@@ -141,10 +141,6 @@ typedef struct STscObj {
SAppInstInfo* pAppInfo;
} STscObj;
typedef struct SMqConsumer {
STscObj* pTscObj;
} SMqConsumer;
typedef struct SReqResultInfo {
const char* pRspMsg;
const char* pData;
......@@ -171,8 +167,8 @@ typedef struct SRequestSendRecvBody {
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.
int64_t queryJob; // query job, created according to sql query DAG.
struct SQueryDag* pDag; // the query dag, generated according to the sql statement.
SReqResultInfo resInfo;
} SRequestSendRecvBody;
......
source/client/inc/clientLog.h
浏览文件 @ 78513e2e
......@@ -22,13 +22,13 @@ extern "C" {
#include "tlog.h"
#define tscFatal(...) do { if (cDebugFlag & DEBUG_FATAL) { taosPrintLog("TSC FATAL ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscError(...) do { if (cDebugFlag & DEBUG_ERROR) { taosPrintLog("TSC ERROR ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscWarn(...) do { if (cDebugFlag & DEBUG_WARN) { taosPrintLog("TSC WARN ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscInfo(...) do { if (cDebugFlag & DEBUG_INFO) { taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscDebug(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscTrace(...) do { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscDebugL(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC ", cDebugFlag, __VA_ARGS__); }} while(0)
#define tscFatal(...) do { if (cDebugFlag & DEBUG_FATAL) { taosPrintLog("TSC FATAL ", DEBUG_FATAL, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscError(...) do { if (cDebugFlag & DEBUG_ERROR) { taosPrintLog("TSC ERROR ", DEBUG_ERROR, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscWarn(...) do { if (cDebugFlag & DEBUG_WARN) { taosPrintLog("TSC WARN ", DEBUG_WARN, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscInfo(...) do { if (cDebugFlag & DEBUG_INFO) { taosPrintLog("TSC ", DEBUG_INFO, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscDebug(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscTrace(...) do { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC ", DEBUG_TRACE, cDebugFlag, __VA_ARGS__); }} while(0)
#define tscDebugL(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
#ifdef __cplusplus
}
......
source/client/src/clientHb.c
浏览文件 @ 78513e2e
......@@ -23,7 +23,7 @@ static SClientHbMgr clientHbMgr = {0};
static int32_t hbCreateThread();
static void hbStopThread();
static int32_t hbMqHbRspHandle(struct SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { return 0; }
static int32_t hbMqHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { return 0; }
static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog *pCatalog) {
int32_t code = 0;
......@@ -104,7 +104,7 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
return TSDB_CODE_SUCCESS;
}
static int32_t hbQueryHbRspHandle(struct SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey));
if (NULL == info) {
tscWarn("fail to get connInfo, may be dropped, connId:%d, type:%d", pRsp->connKey.connId, pRsp->connKey.hbType);
......@@ -163,7 +163,7 @@ static int32_t hbQueryHbRspHandle(struct SAppHbMgr *pAppHbMgr, SClientHbRsp *pRs
return TSDB_CODE_SUCCESS;
}
static int32_t hbMqAsyncCallBack(void *param, const SDataBuf *pMsg, int32_t code) {
static int32_t hbAsyncCallBack(void *param, const SDataBuf *pMsg, int32_t code) {
static int32_t emptyRspNum = 0;
if (code != 0) {
tfree(param);
......@@ -226,7 +226,11 @@ int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SCl
db->vgVersion = htonl(db->vgVersion);
}
SKv kv = {.key = HEARTBEAT_KEY_DBINFO, .valueLen = sizeof(SDbVgVersion) * dbNum, .value = dbs};
SKv kv = {
.key = HEARTBEAT_KEY_DBINFO,
.valueLen = sizeof(SDbVgVersion) * dbNum,
.value = dbs,
};
tscDebug("hb got %d expired db, valueLen:%d", dbNum, kv.valueLen);
......@@ -256,7 +260,11 @@ int32_t hbGetExpiredStbInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SC
stb->tversion = htons(stb->tversion);
}
SKv kv = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = sizeof(SSTableMetaVersion) * stbNum, .value = stbs};
SKv kv = {
.key = HEARTBEAT_KEY_STBINFO,
.valueLen = sizeof(SSTableMetaVersion) * stbNum,
.value = stbs,
};
tscDebug("hb got %d expired stb, valueLen:%d", stbNum, kv.valueLen);
......@@ -288,7 +296,7 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
return TSDB_CODE_SUCCESS;
}
int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) {}
int32_t hbMqHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req) { return 0; }
void hbMgrInitMqHbHandle() {
clientHbMgr.reqHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbReqHandle;
......@@ -396,7 +404,7 @@ static void *hbThreadFunc(void *param) {
free(buf);
break;
}
pInfo->fp = hbMqAsyncCallBack;
pInfo->fp = hbAsyncCallBack;
pInfo->msgInfo.pData = buf;
pInfo->msgInfo.len = tlen;
pInfo->msgType = TDMT_MND_HEARTBEAT;
......@@ -448,7 +456,6 @@ static void hbStopThread() {
}
SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) {
/*return NULL;*/
hbMgrInit();
SAppHbMgr *pAppHbMgr = malloc(sizeof(SAppHbMgr));
if (pAppHbMgr == NULL) {
......@@ -473,7 +480,8 @@ SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) {
free(pAppHbMgr);
return NULL;
}
pAppHbMgr->activeInfo->freeFp = tFreeClientHbReq;
taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
// init getInfoFunc
pAppHbMgr->connInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
......@@ -506,7 +514,6 @@ void appHbMgrCleanup(void) {
}
int hbMgrInit() {
/*return 0;*/
// init once
int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 0, 1);
if (old == 1) return 0;
......@@ -524,7 +531,7 @@ int hbMgrInit() {
}
void hbMgrCleanUp() {
return;
#if 0
hbStopThread();
// destroy all appHbMgr
......@@ -537,6 +544,7 @@ void hbMgrCleanUp() {
pthread_mutex_unlock(&clientHbMgr.lock);
clientHbMgr.appHbMgrs = NULL;
#endif
}
int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *info) {
......@@ -563,9 +571,11 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *
}
int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType) {
/*return 0;*/
SClientHbKey connKey = {.connId = connId, .hbType = HEARTBEAT_TYPE_QUERY};
SHbConnInfo info = {0};
SClientHbKey connKey = {
.connId = connId,
.hbType = HEARTBEAT_TYPE_QUERY,
};
SHbConnInfo info = {0};
switch (hbType) {
case HEARTBEAT_TYPE_QUERY: {
......@@ -586,7 +596,6 @@ int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int3
}
void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) {
/*return;*/
int32_t code = 0;
code = taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
code = taosHashRemove(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey));
......@@ -598,7 +607,6 @@ void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) {
int hbAddConnInfo(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, void *key, void *value, int32_t keyLen,
int32_t valueLen) {
return 0;
// find req by connection id
SClientHbReq *pReq = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
ASSERT(pReq != NULL);
......
source/client/src/clientImpl.c
浏览文件 @ 78513e2e
......@@ -4,8 +4,8 @@
#include "parser.h"
#include "planner.h"
#include "scheduler.h"
#include "tdatablock.h"
#include "tdef.h"
#include "tep.h"
#include "tglobal.h"
#include "tmsgtype.h"
#include "tpagedbuf.h"
......@@ -227,10 +227,10 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList) {
void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.pQueryJob, pRequest->sqlstr, &res);
int32_t code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr, &res);
if (code != TSDB_CODE_SUCCESS) {
if (pRequest->body.pQueryJob != NULL) {
schedulerFreeJob(pRequest->body.pQueryJob);
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
}
pRequest->code = code;
......@@ -240,8 +240,8 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, SArray* pNodeList)
if (TSDB_SQL_INSERT == pRequest->type || TSDB_SQL_CREATE_TABLE == pRequest->type) {
pRequest->body.resInfo.numOfRows = res.numOfRows;
if (pRequest->body.pQueryJob != NULL) {
schedulerFreeJob(pRequest->body.pQueryJob);
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
}
}
......@@ -494,7 +494,7 @@ void* doFetchRow(SRequestObj* pRequest) {
}
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
int32_t code = schedulerFetchRows(pRequest->body.pQueryJob, (void**)&pResInfo->pData);
int32_t code = schedulerFetchRows(pRequest->body.queryJob, (void**)&pResInfo->pData);
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
return NULL;
......
source/client/src/tmq.c
浏览文件 @ 78513e2e
此差异已折叠。
source/common/src/tep.c → source/common/src/tdatablock.c
浏览文件 @ 78513e2e
......@@ -14,7 +14,7 @@
*/
#define _DEFAULT_SOURCE
#include "tep.h"
#include "tdatablock.h"
#include "tcompare.h"
#include "tglobal.h"
......
source/common/src/tglobal.c
浏览文件 @ 78513e2e
......@@ -17,7 +17,7 @@
#include "tglobal.h"
#include "tcompare.h"
#include "tconfig.h"
#include "tep.h"
#include "tdatablock.h"
#include "tlog.h"
SConfig *tsCfg = NULL;
......@@ -155,21 +155,40 @@ static void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t prima
uTrace("dataDir:%s, level:%d primary:%d is configured", v1, level, primary);
}
static void taosSetTfsCfg(SConfig *pCfg) {
static int32_t taosSetTfsCfg(SConfig *pCfg) {
SConfigItem *pItem = cfgGetItem(pCfg, "dataDir");
if (pItem == NULL) return;
memset(tsDataDir, 0, PATH_MAX);
int32_t size = taosArrayGetSize(pItem->array);
if (size <= 0) {
tsDiskCfgNum = 1;
taosAddDataDir(0, pItem->str, 0, 1);
tstrncpy(tsDataDir, pItem->str, PATH_MAX);
if (taosMkDir(tsDataDir) != 0) {
uError("failed to create dataDir:%s since %s", tsDataDir, terrstr());
return -1;
}
} else {
tsDiskCfgNum = size < TFS_MAX_DISKS ? size : TFS_MAX_DISKS;
for (int32_t index = 0; index < tsDiskCfgNum; ++index) {
SDiskCfg *pCfg = taosArrayGet(pItem->array, index);
memcpy(&tsDiskCfg[index], pCfg, sizeof(SDiskCfg));
if (pCfg->level == 0 && pCfg->primary == 1) {
tstrncpy(tsDataDir, pCfg->dir, PATH_MAX);
}
if (taosMkDir(pCfg->dir) != 0) {
uError("failed to create tfsDir:%s since %s", tsDataDir, terrstr());
return -1;
}
}
}
if (tsDataDir[0] == 0) {
uError("datadir not set");
return -1;
}
return 0;
}
struct SConfig *taosGetCfg() {
......@@ -279,11 +298,11 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
if (cfgAddLocale(pCfg, "locale", tsLocale) != 0) return -1;
if (cfgAddCharset(pCfg, "charset", tsCharset) != 0) return -1;
if (cfgAddBool(pCfg, "enableCoreFile", 1, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "numOfCores", tsNumOfCores, 1, 100000, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "pageSize(KB)", tsPageSize, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 0, 100000, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "openMax", tsOpenMax, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "streamMax", tsStreamMax, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "totalMemory(MB)", tsTotalMemoryMB, 0, INT32_MAX, 1) != 0) return -1;
if (cfgAddInt32(pCfg, "pageSize(KB)", tsPageSizeKB, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddInt64(pCfg, "totalMemory(KB)", tsTotalMemoryKB, 0, INT64_MAX, 1) != 0) return -1;
if (cfgAddString(pCfg, "os sysname", info.sysname, 1) != 0) return -1;
if (cfgAddString(pCfg, "os nodename", info.nodename, 1) != 0) return -1;
if (cfgAddString(pCfg, "os release", info.release, 1) != 0) return -1;
......@@ -358,7 +377,7 @@ static void taosSetServerLogCfg(SConfig *pCfg) {
fsDebugFlag = cfgGetItem(pCfg, "fsDebugFlag")->i32;
}
static void taosSetClientCfg(SConfig *pCfg) {
static int32_t taosSetClientCfg(SConfig *pCfg) {
tstrncpy(tsLocalFqdn, cfgGetItem(pCfg, "fqdn")->str, TSDB_FQDN_LEN);
tsServerPort = (uint16_t)cfgGetItem(pCfg, "serverPort")->i32;
snprintf(tsLocalEp, sizeof(tsLocalEp), "%s:%u", tsLocalFqdn, tsServerPort);
......@@ -375,9 +394,13 @@ static void taosSetClientCfg(SConfig *pCfg) {
snprintf(tsSecond, sizeof(tsSecond), "%s:%u", secondEp.fqdn, secondEp.port);
cfgSetItem(pCfg, "secondEp", tsSecond, pSecondpItem->stype);
tstrncpy(tsLogDir, cfgGetItem(pCfg, "tempDir")->str, PATH_MAX);
taosExpandDir(tsLogDir, tsLogDir, PATH_MAX);
tstrncpy(tsTempDir, cfgGetItem(pCfg, "tempDir")->str, PATH_MAX);
taosExpandDir(tsTempDir, tsTempDir, PATH_MAX);
tsTempSpace.reserved = cfgGetItem(pCfg, "minimalTempDirGB")->fval;
if (taosMkDir(tsTempDir) != 0) {
uError("failed to create tempDir:%s since %s", tsTempDir, terrstr());
return -1;
}
tsNumOfThreadsPerCore = cfgGetItem(pCfg, "maxTmrCtrl")->fval;
tsMaxTmrCtrl = cfgGetItem(pCfg, "maxTmrCtrl")->i32;
......@@ -392,6 +415,8 @@ static void taosSetClientCfg(SConfig *pCfg) {
tsMaxNumOfOrderedResults = cfgGetItem(pCfg, "maxNumOfOrderedRes")->i32;
tsKeepOriginalColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
tsMaxBinaryDisplayWidth = cfgGetItem(pCfg, "maxBinaryDisplayWidth")->i32;
return 0;
}
static void taosSetSystemCfg(SConfig *pCfg) {
......@@ -404,10 +429,6 @@ static void taosSetSystemCfg(SConfig *pCfg) {
const char *charset = cfgGetItem(pCfg, "charset")->str;
taosSetSystemLocale(locale, charset);
if (tsNumOfCores <= 1) {
tsNumOfCores = 2;
}
bool enableCore = cfgGetItem(pCfg, "enableCoreFile")->bval;
taosSetConsoleEcho(enableCore);
......@@ -415,11 +436,8 @@ static void taosSetSystemCfg(SConfig *pCfg) {
tsVersion = 30000000;
}
static void taosSetServerCfg(SConfig *pCfg) {
tstrncpy(tsDataDir, cfgGetItem(pCfg, "dataDir")->str, PATH_MAX);
taosExpandDir(tsDataDir, tsDataDir, PATH_MAX);
tsTempSpace.reserved = cfgGetItem(pCfg, "minimalDataDirGB")->fval;
static int32_t taosSetServerCfg(SConfig *pCfg) {
tsDataSpace.reserved = cfgGetItem(pCfg, "minimalDataDirGB")->fval;
tsNumOfCommitThreads = cfgGetItem(pCfg, "numOfCommitThreads")->i32;
tsRatioOfQueryCores = cfgGetItem(pCfg, "ratioOfQueryCores")->fval;
tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
......@@ -448,6 +466,8 @@ static void taosSetServerCfg(SConfig *pCfg) {
if (tsQueryBufferSize >= 0) {
tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
}
return 0;
}
int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char *envFile,
......@@ -458,10 +478,10 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
if (pCfg == NULL) return -1;
if (tsc) {
tscEmbeddedInUtil = 0;
tsLogEmbedded = 0;
if (taosAddClientLogCfg(pCfg) != 0) return -1;
} else {
tscEmbeddedInUtil = 1;
tsLogEmbedded = 1;
if (taosAddClientLogCfg(pCfg) != 0) return -1;
if (taosAddServerLogCfg(pCfg) != 0) return -1;
}
......@@ -508,8 +528,8 @@ int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloU
tsCfg = cfgInit();
if (tsc) {
if (taosAddClientLogCfg(tsCfg) != 0) return -1;
if (taosAddClientCfg(tsCfg) != 0) return -1;
if (taosAddClientLogCfg(tsCfg) != 0) return -1;
} else {
if (taosAddClientCfg(tsCfg) != 0) return -1;
if (taosAddServerCfg(tsCfg) != 0) return -1;
......@@ -532,24 +552,14 @@ int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloU
}
if (tsc) {
taosSetClientCfg(tsCfg);
if (taosSetClientCfg(tsCfg)) return -1;
} else {
taosSetClientCfg(tsCfg);
taosSetServerCfg(tsCfg);
taosSetTfsCfg(tsCfg);
if (taosSetClientCfg(tsCfg)) return -1;
if (taosSetServerCfg(tsCfg)) return -1;
if (taosSetTfsCfg(tsCfg) != 0) return -1;
}
taosSetSystemCfg(tsCfg);
if (taosMkDir(tsTempDir) != 0) {
uError("failed to create dir:%s since %s", tsTempDir, terrstr());
return -1;
}
if (!tsc && taosMkDir(tsDataDir) != 0) {
uError("failed to create dir:%s since %s", tsDataDir, terrstr());
return -1;
}
cfgDumpCfg(tsCfg, tsc, false);
return 0;
}
......
source/common/src/tmsg.c
浏览文件 @ 78513e2e
......@@ -2112,6 +2112,9 @@ int32_t tSerializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *pR
if (tEncodeI32(&encoder, pReq->maxRows) < 0) return -1;
if (tEncodeI32(&encoder, pReq->commitTime) < 0) return -1;
if (tEncodeI32(&encoder, pReq->fsyncPeriod) < 0) return -1;
if (tEncodeU32(&encoder, pReq->hashBegin) < 0) return -1;
if (tEncodeU32(&encoder, pReq->hashEnd) < 0) return -1;
if (tEncodeI8(&encoder, pReq->hashMethod) < 0) return -1;
if (tEncodeI8(&encoder, pReq->walLevel) < 0) return -1;
if (tEncodeI8(&encoder, pReq->precision) < 0) return -1;
if (tEncodeI8(&encoder, pReq->compression) < 0) return -1;
......@@ -2152,6 +2155,9 @@ int32_t tDeserializeSCreateVnodeReq(void *buf, int32_t bufLen, SCreateVnodeReq *
if (tDecodeI32(&decoder, &pReq->maxRows) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->commitTime) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->fsyncPeriod) < 0) return -1;
if (tDecodeU32(&decoder, &pReq->hashBegin) < 0) return -1;
if (tDecodeU32(&decoder, &pReq->hashEnd) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->hashMethod) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->walLevel) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->precision) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->compression) < 0) return -1;
......
source/common/src/ttszip.c
浏览文件 @ 78513e2e
......@@ -46,7 +46,7 @@ STSBuf* tsBufCreate(bool autoDelete, int32_t order) {
}
if (!autoDelete) {
remove(pTSBuf->path);
taosRemoveFile(pTSBuf->path);
}
if (NULL == allocResForTSBuf(pTSBuf)) {
......@@ -178,7 +178,7 @@ void* tsBufDestroy(STSBuf* pTSBuf) {
if (pTSBuf->autoDelete) {
// ("tsBuf %p destroyed, delete tmp file:%s", pTSBuf, pTSBuf->path);
remove(pTSBuf->path);
taosRemoveFile(pTSBuf->path);
} else {
// tscDebug("tsBuf %p destroyed, tmp file:%s, remains", pTSBuf, pTSBuf->path);
}
......
source/common/test/commonTests.cpp
浏览文件 @ 78513e2e
#include "tcommon.h"
#include <gtest/gtest.h>
#include <tep.h>
#include <iostream>
#pragma GCC diagnostic push
......@@ -10,6 +8,9 @@
#pragma GCC diagnostic ignored "-Wsign-compare"
#include "os.h"
#include "tcommon.h"
#include "tdatablock.h"
#include "tcommon.h"
#include "taos.h"
#include "tvariant.h"
#include "tdef.h"
......
source/common/test/tmsgTest.cpp
浏览文件 @ 78513e2e
#include <iostream>
#include "gtest/gtest.h"
#include <gtest/gtest.h>
#include "tmsg.h"
......
source/dnode/mgmt/impl/inc/dndEnv.h
浏览文件 @ 78513e2e
......@@ -31,7 +31,7 @@ typedef struct {
SDnode *pDnode;
STaosQueue *queue;
union {
SQWorkerPool pool;
SQWorkerPool pool;
SWWorkerPool mpool;
};
} SDnodeWorker;
......@@ -137,6 +137,8 @@ typedef struct SDnode {
SStartupReq startup;
} SDnode;
int32_t dndGetMonitorDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo);
#ifdef __cplusplus
}
#endif
......
source/dnode/mgmt/impl/inc/dndInt.h
浏览文件 @ 78513e2e
......@@ -23,9 +23,11 @@ extern "C" {
#include "os.h"
#include "cJSON.h"
#include "monitor.h"
#include "tcache.h"
#include "tcrc32c.h"
#include "tep.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "thash.h"
#include "tlockfree.h"
#include "tlog.h"
......@@ -35,8 +37,6 @@ extern "C" {
#include "tthread.h"
#include "ttime.h"
#include "tworker.h"
#include "tglobal.h"
#include "monitor.h"
#include "dnode.h"
......@@ -47,14 +47,12 @@ extern "C" {
#include "vnode.h"
#include "tfs.h"
extern int32_t dDebugFlag;
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", 255, __VA_ARGS__); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", 255, __VA_ARGS__); }}
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", 255, __VA_ARGS__); }}
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", 255, __VA_ARGS__); }}
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", dDebugFlag, __VA_ARGS__); }}
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", dDebugFlag, __VA_ARGS__); }}
#define dFatal(...) { if (dDebugFlag & DEBUG_FATAL) { taosPrintLog("DND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
#define dError(...) { if (dDebugFlag & DEBUG_ERROR) { taosPrintLog("DND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
#define dWarn(...) { if (dDebugFlag & DEBUG_WARN) { taosPrintLog("DND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
#define dInfo(...) { if (dDebugFlag & DEBUG_INFO) { taosPrintLog("DND ", DEBUG_INFO, 255, __VA_ARGS__); }}
#define dDebug(...) { if (dDebugFlag & DEBUG_DEBUG) { taosPrintLog("DND ", DEBUG_DEBUG, dDebugFlag, __VA_ARGS__); }}
#define dTrace(...) { if (dDebugFlag & DEBUG_TRACE) { taosPrintLog("DND ", DEBUG_TRACE, dDebugFlag, __VA_ARGS__); }}
typedef enum { DND_STAT_INIT, DND_STAT_RUNNING, DND_STAT_STOPPED } EStat;
typedef enum { DND_WORKER_SINGLE, DND_WORKER_MULTI } EWorkerType;
......
source/dnode/mgmt/impl/inc/dndMnode.h
浏览文件 @ 78513e2e
......@@ -34,6 +34,7 @@ int32_t dndProcessDropMnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
int32_t dndGetMnodeMonitorInfo(SDnode *pDnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo);
int8_t dndIsMnode(SDnode *pDnode);
#ifdef __cplusplus
}
......
source/dnode/mgmt/impl/src/dndEnv.c
浏览文件 @ 78513e2e
......@@ -323,3 +323,12 @@ void dndCleanup() {
taosStopCacheRefreshWorker();
dInfo("dnode env is cleaned up");
}
int32_t dndGetMonitorDiskInfo(SDnode *pDnode, SMonDiskInfo *pInfo) {
tstrncpy(pInfo->logdir.name, tsLogDir, sizeof(pInfo->logdir.name));
pInfo->logdir.size = tsLogSpace.size;
tstrncpy(pInfo->tempdir.name, tsTempDir, sizeof(pInfo->tempdir.name));
pInfo->tempdir.size = tsTempSpace.size;
return tfsGetMonitorInfo(pDnode->pTfs, pInfo);
}
\ No newline at end of file
source/dnode/mgmt/impl/src/dndMgmt.c
浏览文件 @ 78513e2e
......@@ -474,20 +474,48 @@ void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
rpcSendResponse(&rpcRsp);
}
void dndGetBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
static void dndGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
pInfo->dnode_id = dndGetDnodeId(pDnode);
tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
}
static void dndGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
pInfo->uptime = (taosGetTimestampMs() - pDnode->dmgmt.rebootTime) / (86400000.0f);
taosGetCpuUsage(&pInfo->cpu_engine, &pInfo->cpu_system);
pInfo->cpu_cores = tsNumOfCores;
taosGetProcMemory(&pInfo->mem_engine);
taosGetSysMemory(&pInfo->mem_system);
pInfo->mem_total = tsTotalMemoryKB;
pInfo->disk_engine = 0;
pInfo->disk_used = tsDataSpace.size.used / (1024 * 1024 * 1024.0);
pInfo->disk_total = tsDataSpace.size.avail / (1024 * 1024 * 1024.0);
taosGetCardInfo(NULL, &pInfo->net_in, &pInfo->net_out);
taosGetProcIO(&pInfo->io_read, &pInfo->io_write);
pInfo->io_read_disk = 0;
pInfo->io_write_disk = 0;
pInfo->req_select = 0;
pInfo->req_select_rate = 0;
pInfo->req_insert = 0;
pInfo->req_insert_success = 0;
pInfo->req_insert_rate = 0;
pInfo->req_insert_batch = 0;
pInfo->req_insert_batch_success = 0;
pInfo->req_insert_batch_rate = 0;
pInfo->errors = 0;
pInfo->vnodes_num = 0;
pInfo->masters = 0;
pInfo->has_mnode = dndIsMnode(pDnode);
}
static void dndSendMonitorReport(SDnode *pDnode) {
if (!tsEnableMonitor || tsMonitorFqdn[0] == 0) return;
if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return;
dTrace("pDnode:%p, send monitor report to %s:%u", pDnode, tsMonitorFqdn, tsMonitorPort);
SMonInfo *pMonitor = monCreateMonitorInfo();
if (pMonitor == NULL) return;
dTrace("pDnode:%p, send monitor report to %s:%u", pDnode, tsMonitorFqdn, tsMonitorPort);
SMonBasicInfo basicInfo = {0};
dndGetBasicInfo(pDnode, &basicInfo);
dndGetMonitorBasicInfo(pDnode, &basicInfo);
monSetBasicInfo(pMonitor, &basicInfo);
SMonClusterInfo clusterInfo = {0};
......@@ -499,6 +527,20 @@ static void dndSendMonitorReport(SDnode *pDnode) {
monSetGrantInfo(pMonitor, &grantInfo);
}
SMonDnodeInfo dnodeInfo = {0};
dndGetMonitorDnodeInfo(pDnode, &dnodeInfo);
monSetDnodeInfo(pMonitor, &dnodeInfo);
SMonDiskInfo diskInfo = {0};
if (dndGetMonitorDiskInfo(pDnode, &diskInfo) == 0) {
monSetDiskInfo(pMonitor, &diskInfo);
}
taosArrayDestroy(clusterInfo.dnodes);
taosArrayDestroy(clusterInfo.mnodes);
taosArrayDestroy(vgroupInfo.vgroups);
taosArrayDestroy(diskInfo.datadirs);
monSendReport(pMonitor);
monCleanupMonitorInfo(pMonitor);
}
......
source/dnode/mgmt/impl/src/dndMnode.c
浏览文件 @ 78513e2e
......@@ -639,4 +639,11 @@ int32_t dndGetMnodeMonitorInfo(SDnode *pDnode, SMonClusterInfo *pClusterInfo, SM
int32_t code = mndGetMonitorInfo(pMnode, pClusterInfo, pVgroupInfo, pGrantInfo);
dndReleaseMnode(pDnode, pMnode);
return code;
}
int8_t dndIsMnode(SDnode *pDnode) {
SMnode *pMnode = dndAcquireMnode(pDnode);
if (pMnode == NULL) return 0;
dndReleaseMnode(pDnode, pMnode);
return 1;
}
\ No newline at end of file
source/dnode/mgmt/impl/src/dndVnodes.c
浏览文件 @ 78513e2e
......@@ -85,7 +85,7 @@ static SVnodeObj *dndAcquireVnode(SDnode *pDnode, int32_t vgId) {
int32_t refCount = 0;
taosRLockLatch(&pMgmt->latch);
taosHashGetClone(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
if (pVnode == NULL) {
terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
} else {
......@@ -523,6 +523,9 @@ static void dndGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
pCfg->walCfg.rollPeriod = 128;
pCfg->walCfg.segSize = 128;
pCfg->walCfg.vgId = pCreate->vgId;
pCfg->hashBegin = pCreate->hashBegin;
pCfg->hashEnd = pCreate->hashEnd;
pCfg->hashMethod = pCreate->hashMethod;
}
static void dndGenerateWrapperCfg(SDnode *pDnode, SCreateVnodeReq *pCreate, SWrapperCfg *pCfg) {
......
source/dnode/mgmt/impl/test/sut/src/client.cpp
浏览文件 @ 78513e2e
......@@ -14,7 +14,7 @@
*/
#include "sut.h"
#include "tep.h"
#include "tdatablock.h"
static void processClientRsp(void* parent, SRpcMsg* pRsp, SEpSet* pEpSet) {
TestClient* client = (TestClient*)parent;
......
source/dnode/mgmt/impl/test/sut/src/sut.cpp
浏览文件 @ 78513e2e
......@@ -28,7 +28,7 @@ void Testbase::InitLog(const char* path) {
wDebugFlag = 0;
sDebugFlag = 0;
tsdbDebugFlag = 0;
tscEmbeddedInUtil = 1;
tsLogEmbedded = 1;
tsAsyncLog = 0;
taosRemoveDir(path);
......
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ 78513e2e
......@@ -34,46 +34,6 @@
extern "C" {
#endif
extern int32_t mDebugFlag;
// mnode log function
#define mFatal(...) \
{ \
if (mDebugFlag & DEBUG_FATAL) { \
taosPrintLog("MND FATAL ", 255, __VA_ARGS__); \
} \
}
#define mError(...) \
{ \
if (mDebugFlag & DEBUG_ERROR) { \
taosPrintLog("MND ERROR ", 255, __VA_ARGS__); \
} \
}
#define mWarn(...) \
{ \
if (mDebugFlag & DEBUG_WARN) { \
taosPrintLog("MND WARN ", 255, __VA_ARGS__); \
} \
}
#define mInfo(...) \
{ \
if (mDebugFlag & DEBUG_INFO) { \
taosPrintLog("MND ", 255, __VA_ARGS__); \
} \
}
#define mDebug(...) \
{ \
if (mDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); \
} \
}
#define mTrace(...) \
{ \
if (mDebugFlag & DEBUG_TRACE) { \
taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); \
} \
}
typedef enum {
MND_AUTH_ACCT_START = 0,
MND_AUTH_ACCT_USER,
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ 78513e2e
......@@ -20,7 +20,7 @@
#include "sdb.h"
#include "tcache.h"
#include "tep.h"
#include "tdatablock.h"
#include "tglobal.h"
#include "tqueue.h"
#include "ttime.h"
......@@ -31,6 +31,13 @@
extern "C" {
#endif
#define mFatal(...) { if (mDebugFlag & DEBUG_FATAL) { taosPrintLog("MND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
#define mError(...) { if (mDebugFlag & DEBUG_ERROR) { taosPrintLog("MND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
#define mWarn(...) { if (mDebugFlag & DEBUG_WARN) { taosPrintLog("MND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
#define mInfo(...) { if (mDebugFlag & DEBUG_INFO) { taosPrintLog("MND ", DEBUG_INFO, 255, __VA_ARGS__); }}
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
typedef int32_t (*MndMsgFp)(SMnodeMsg *pMsg);
typedef int32_t (*MndInitFp)(SMnode *pMnode);
typedef void (*MndCleanupFp)(SMnode *pMnode);
......@@ -85,6 +92,11 @@ typedef struct {
ESyncState state;
} SSyncMgmt;
typedef struct {
int64_t expireTimeMS;
int64_t timeseriesAllowed;
} SGrantInfo;
typedef struct SMnode {
int32_t dnodeId;
int64_t clusterId;
......@@ -105,6 +117,7 @@ typedef struct SMnode {
STelemMgmt telemMgmt;
SSyncMgmt syncMgmt;
SHashObj *infosMeta;
SGrantInfo grant;
MndMsgFp msgFp[TDMT_MAX];
SendReqToDnodeFp sendReqToDnodeFp;
SendReqToMnodeFp sendReqToMnodeFp;
......@@ -120,7 +133,7 @@ void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp);
uint64_t mndGenerateUid(char *name, int32_t len);
int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
void mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
#ifdef __cplusplus
}
......
source/dnode/mnode/impl/inc/mndProfile.h
浏览文件 @ 78513e2e
......@@ -24,6 +24,7 @@ extern "C" {
int32_t mndInitProfile(SMnode *pMnode);
void mndCleanupProfile(SMnode *pMnode);
int32_t mndGetNumOfConnections(SMnode *pMnode);
#ifdef __cplusplus
}
......
source/dnode/mnode/impl/src/mndMnode.c
浏览文件 @ 78513e2e
......@@ -99,11 +99,15 @@ void mndUpdateMnodeRole(SMnode *pMnode) {
pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);
if (pIter == NULL) break;
ESyncState lastRole = pObj->role;
if (pObj->id == 1) {
pObj->role = TAOS_SYNC_STATE_LEADER;
} else {
pObj->role = TAOS_SYNC_STATE_CANDIDATE;
}
if (pObj->role != lastRole) {
pObj->roleTime = taosGetTimestampMs();
}
sdbRelease(pSdb, pObj);
}
......
source/dnode/mnode/impl/src/mndProfile.c
浏览文件 @ 78513e2e
......@@ -48,7 +48,7 @@ static SConnObj *mndCreateConn(SMnode *pMnode, SRpcConnInfo *pInfo, int32_t pid,
static void mndFreeConn(SConnObj *pConn);
static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId);
static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn);
static void *mndGetNextConn(SMnode *pMnode, void *pIter, SConnObj **pConn);
static void *mndGetNextConn(SMnode *pMnode, SCacheIter *pIter);
static void mndCancelGetNextConn(SMnode *pMnode, void *pIter);
static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq);
static int32_t mndProcessConnectReq(SMnodeMsg *pReq);
......@@ -158,27 +158,23 @@ static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn) {
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
}
static void *mndGetNextConn(SMnode *pMnode, void *pIter, SConnObj **pConn) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
*pConn = NULL;
pIter = taosHashIterate(pMgmt->cache->pHashTable, pIter);
if (pIter == NULL) return NULL;
SCacheDataNode **pNode = pIter;
if (pNode == NULL || *pNode == NULL) {
taosHashCancelIterate(pMgmt->cache->pHashTable, pIter);
return NULL;
void *mndGetNextConn(SMnode *pMnode, SCacheIter *pIter) {
SConnObj* pConn = NULL;
bool hasNext = taosCacheIterNext(pIter);
if (hasNext) {
size_t dataLen = 0;
pConn = taosCacheIterGetData(pIter, &dataLen);
} else {
taosCacheDestroyIter(pIter);
}
*pConn = (SConnObj *)((*pNode)->data);
return pIter;
return pConn;
}
static void mndCancelGetNextConn(SMnode *pMnode, void *pIter) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
taosHashCancelIterate(pMgmt->cache->pHashTable, pIter);
if (pIter != NULL) {
taosCacheDestroyIter(pIter);
}
}
static int32_t mndProcessConnectReq(SMnodeMsg *pReq) {
......@@ -376,8 +372,8 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
int32_t rspLen = 0;
mndValidateDbInfo(pMnode, kv->value, kv->valueLen / sizeof(SDbVgVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) {
SKv kv = {.key = HEARTBEAT_KEY_DBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv);
SKv kv1 = {.key = HEARTBEAT_KEY_DBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1);
}
break;
}
......@@ -386,8 +382,8 @@ static int32_t mndProcessHeartBeatReq(SMnodeMsg *pReq) {
int32_t rspLen = 0;
mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableMetaVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) {
SKv kv = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv);
SKv kv1 = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1);
}
break;
}
......@@ -638,7 +634,7 @@ static int32_t mndGetConnsMeta(SMnodeMsg *pReq, SShowObj *pShow, STableMetaRsp *
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = taosHashGetSize(pMgmt->cache->pHashTable);
pShow->numOfRows = taosCacheGetNumOfObj(pMgmt->cache);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
......@@ -653,8 +649,13 @@ static int32_t mndRetrieveConns(SMnodeMsg *pReq, SShowObj *pShow, char *data, in
char *pWrite;
char ipStr[TSDB_IPv4ADDR_LEN + 6];
if (pShow->pIter == NULL) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
pShow->pIter = taosCacheCreateIter(pMgmt->cache);
}
while (numOfRows < rows) {
pShow->pIter = mndGetNextConn(pMnode, pShow->pIter, &pConn);
pConn = mndGetNextConn(pMnode, pShow->pIter);
if (pConn == NULL) break;
cols = 0;
......@@ -823,19 +824,24 @@ static int32_t mndRetrieveQueries(SMnodeMsg *pReq, SShowObj *pShow, char *data,
void *pIter;
char str[TSDB_IPv4ADDR_LEN + 6] = {0};
if (pShow->pIter == NULL) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
pShow->pIter = taosCacheCreateIter(pMgmt->cache);
}
while (numOfRows < rows) {
pIter = mndGetNextConn(pMnode, pShow->pIter, &pConn);
pConn = mndGetNextConn(pMnode, pShow->pIter);
if (pConn == NULL) {
pShow->pIter = pIter;
pShow->pIter = NULL;
break;
}
if (numOfRows + pConn->numOfQueries >= rows) {
mndCancelGetNextConn(pMnode, pIter);
taosCacheDestroyIter(pShow->pIter);
pShow->pIter = NULL;
break;
}
pShow->pIter = pIter;
for (int32_t i = 0; i < pConn->numOfQueries; ++i) {
SQueryDesc *pDesc = pConn->pQueries + i;
cols = 0;
......@@ -913,6 +919,12 @@ static int32_t mndRetrieveQueries(SMnodeMsg *pReq, SShowObj *pShow, char *data,
}
static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
taosHashCancelIterate(pMgmt->cache->pHashTable, pIter);
if (pIter != NULL) {
taosCacheDestroyIter(pIter);
}
}
int32_t mndGetNumOfConnections(SMnode *pMnode) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
return taosCacheGetNumOfObj(pMgmt->cache);
}
\ No newline at end of file
source/dnode/mnode/impl/src/mndTelem.c
浏览文件 @ 78513e2e
......@@ -18,15 +18,15 @@
#include "mndCluster.h"
#include "mndSync.h"
#include "tbuffer.h"
#include "tjson.h"
#include "thttp.h"
#include "tjson.h"
#define TELEMETRY_SERVER "telemetry.taosdata.com"
#define TELEMETRY_PORT 80
static void mndBuildRuntimeInfo(SMnode* pMnode, SJson* pJson) {
SMnodeLoad load = {0};
if (mndGetLoad(pMnode, &load) != 0) return;
mndGetLoad(pMnode, &load);
tjsonAddDoubleToObject(pJson, "numOfDnode", load.numOfDnode);
tjsonAddDoubleToObject(pJson, "numOfMnode", load.numOfMnode);
......@@ -52,23 +52,20 @@ static char* mndBuildTelemetryReport(SMnode* pMnode) {
tjsonAddStringToObject(pJson, "instanceId", clusterName);
tjsonAddDoubleToObject(pJson, "reportVersion", 1);
if (taosGetOsReleaseName(tmp, sizeof(tmp))) {
if (taosGetOsReleaseName(tmp, sizeof(tmp)) == 0) {
tjsonAddStringToObject(pJson, "os", tmp);
}
int32_t numOfCores = 0;
if (taosGetCpuInfo(tmp, sizeof(tmp), &numOfCores)) {
float numOfCores = 0;
if (taosGetCpuInfo(tmp, sizeof(tmp), &numOfCores) == 0) {
tjsonAddStringToObject(pJson, "cpuModel", tmp);
tjsonAddDoubleToObject(pJson, "numOfCpu", numOfCores);
} else {
tjsonAddDoubleToObject(pJson, "numOfCpu", taosGetCpuCores());
tjsonAddDoubleToObject(pJson, "numOfCpu", tsNumOfCores);
}
uint64_t memoryKB = 0;
if (taosGetTotalSysMemoryKB(&memoryKB)) {
snprintf(tmp, sizeof(tmp), "%" PRIu64 " kB", memoryKB);
tjsonAddStringToObject(pJson, "memory", tmp);
}
snprintf(tmp, sizeof(tmp), "%" PRId64 " kB", tsTotalMemoryKB);
tjsonAddStringToObject(pJson, "memory", tmp);
tjsonAddStringToObject(pJson, "version", version);
tjsonAddStringToObject(pJson, "buildInfo", buildinfo);
......
source/dnode/mnode/impl/src/mndVgroup.c
浏览文件 @ 78513e2e
......@@ -215,6 +215,9 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
createReq.replica = pVgroup->replica;
createReq.selfIndex = -1;
createReq.streamMode = pVgroup->streamMode;
createReq.hashBegin = pVgroup->hashBegin;
createReq.hashEnd = pVgroup->hashEnd;
createReq.hashMethod = pDb->hashMethod;
for (int32_t v = 0; v < pVgroup->replica; ++v) {
SReplica *pReplica = &createReq.replicas[v];
......
source/dnode/mnode/impl/src/mnode.c
浏览文件 @ 78513e2e
......@@ -359,6 +359,7 @@ SMnode *mndOpen(const char *path, const SMnodeOpt *pOption) {
return NULL;
}
mndUpdateMnodeRole(pMnode);
mDebug("mnode open successfully ");
return pMnode;
}
......@@ -385,26 +386,6 @@ void mndDestroy(const char *path) {
mDebug("mnode is destroyed");
}
int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
pLoad->numOfDnode = 0;
pLoad->numOfMnode = 0;
pLoad->numOfVgroup = 0;
pLoad->numOfDatabase = 0;
pLoad->numOfSuperTable = 0;
pLoad->numOfChildTable = 0;
pLoad->numOfColumn = 0;
pLoad->totalPoints = 0;
pLoad->totalStorage = 0;
pLoad->compStorage = 0;
return 0;
}
int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo) {
return 0;
}
SMnodeMsg *mndInitMsg(SMnode *pMnode, SRpcMsg *pRpcMsg) {
SMnodeMsg *pMsg = taosAllocateQitem(sizeof(SMnodeMsg));
if (pMsg == NULL) {
......@@ -523,3 +504,131 @@ uint64_t mndGenerateUid(char *name, int32_t len) {
}
} while (true);
}
void mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
memset(pLoad, 0, sizeof(SMnodeLoad));
SSdb *pSdb = pMnode->pSdb;
pLoad->numOfDnode = sdbGetSize(pSdb, SDB_DNODE);
pLoad->numOfMnode = sdbGetSize(pSdb, SDB_MNODE);
pLoad->numOfVgroup = sdbGetSize(pSdb, SDB_VGROUP);
pLoad->numOfDatabase = sdbGetSize(pSdb, SDB_DB);
pLoad->numOfSuperTable = sdbGetSize(pSdb, SDB_STB);
void *pIter = NULL;
while (1) {
SVgObj *pVgroup = NULL;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
pLoad->numOfChildTable += pVgroup->numOfTables;
pLoad->numOfColumn += pVgroup->numOfTimeSeries;
pLoad->totalPoints += pVgroup->pointsWritten;
pLoad->totalStorage += pVgroup->totalStorage;
pLoad->compStorage += pVgroup->compStorage;
sdbRelease(pSdb, pVgroup);
}
}
int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
SMonGrantInfo *pGrantInfo) {
if (!mndIsMaster(pMnode)) return -1;
SSdb *pSdb = pMnode->pSdb;
int64_t ms = taosGetTimestampMs();
pClusterInfo->dnodes = taosArrayInit(sdbGetSize(pSdb, SDB_DNODE), sizeof(SMonDnodeDesc));
pClusterInfo->mnodes = taosArrayInit(sdbGetSize(pSdb, SDB_MNODE), sizeof(SMonMnodeDesc));
pVgroupInfo->vgroups = taosArrayInit(sdbGetSize(pSdb, SDB_VGROUP), sizeof(SMonVgroupDesc));
if (pClusterInfo->dnodes == NULL || pClusterInfo->mnodes == NULL || pVgroupInfo->vgroups == NULL) {
return -1;
}
// cluster info
tstrncpy(pClusterInfo->version, version, sizeof(pClusterInfo->version));
pClusterInfo->monitor_interval = tsMonitorInterval;
pClusterInfo->connections_total = mndGetNumOfConnections(pMnode);
void *pIter = NULL;
while (1) {
SDnodeObj *pObj = NULL;
pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pObj);
if (pIter == NULL) break;
SMonDnodeDesc desc = {0};
desc.dnode_id = pObj->id;
tstrncpy(desc.dnode_ep, pObj->ep, sizeof(desc.dnode_ep));
if (mndIsDnodeOnline(pMnode, pObj, ms)) {
tstrncpy(desc.status, "ready", sizeof(desc.status));
} else {
tstrncpy(desc.status, "offline", sizeof(desc.status));
}
taosArrayPush(pClusterInfo->dnodes, &desc);
sdbRelease(pSdb, pObj);
}
pIter = NULL;
while (1) {
SMnodeObj *pObj = NULL;
pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);
if (pIter == NULL) break;
SMonMnodeDesc desc = {0};
desc.mnode_id = pObj->id;
tstrncpy(desc.mnode_ep, pObj->pDnode->ep, sizeof(desc.mnode_ep));
tstrncpy(desc.role, mndGetRoleStr(pObj->role), sizeof(desc.role));
taosArrayPush(pClusterInfo->mnodes, &desc);
sdbRelease(pSdb, pObj);
if (pObj->role == TAOS_SYNC_STATE_LEADER) {
pClusterInfo->first_ep_dnode_id = pObj->id;
tstrncpy(pClusterInfo->first_ep, pObj->pDnode->ep, sizeof(pClusterInfo->first_ep));
pClusterInfo->master_uptime = (ms - pObj->roleTime) / (86400000.0f);
}
}
// vgroup info
pIter = NULL;
while (1) {
SVgObj *pVgroup = NULL;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
pClusterInfo->vgroups_total++;
SMonVgroupDesc desc = {0};
desc.vgroup_id = pVgroup->vgId;
strncpy(desc.database_name, pVgroup->dbName, sizeof(desc.database_name));
desc.tables_num = pVgroup->numOfTables;
pGrantInfo->timeseries_used += pVgroup->numOfTimeSeries;
tstrncpy(desc.status, "unsynced", sizeof(desc.status));
for (int32_t i = 0; i < pVgroup->replica; ++i) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
SMonVnodeDesc *pVnDesc = &desc.vnodes[i];
pVnDesc->dnode_id = pVgid->dnodeId;
tstrncpy(pVnDesc->vnode_role, mndGetRoleStr(pVgid->role), sizeof(pVnDesc->vnode_role));
if (pVgid->role == TAOS_SYNC_STATE_LEADER) {
tstrncpy(desc.status, "ready", sizeof(desc.status));
pClusterInfo->vgroups_alive++;
}
if (pVgid->role == TAOS_SYNC_STATE_LEADER || pVgid->role == TAOS_SYNC_STATE_CANDIDATE) {
pClusterInfo->vnodes_alive++;
}
pClusterInfo->vnodes_total++;
}
taosArrayPush(pVgroupInfo->vgroups, &desc);
sdbRelease(pSdb, pVgroup);
}
// grant info
pGrantInfo->expire_time = (pMnode->grant.expireTimeMS - ms) / 86400000.0f;
pGrantInfo->timeseries_total = pMnode->grant.timeseriesAllowed;
if (pMnode->grant.expireTimeMS == 0) {
pGrantInfo->expire_time = INT32_MAX;
pGrantInfo->timeseries_total = INT32_MAX;
}
return 0;
}
\ No newline at end of file
source/dnode/mnode/sdb/inc/sdbInt.h
浏览文件 @ 78513e2e
......@@ -28,12 +28,12 @@
extern "C" {
#endif
#define mFatal(...) { if (mDebugFlag & DEBUG_FATAL) { taosPrintLog("MND FATAL ", 255, __VA_ARGS__); }}
#define mError(...) { if (mDebugFlag & DEBUG_ERROR) { taosPrintLog("MND ERROR ", 255, __VA_ARGS__); }}
#define mWarn(...) { if (mDebugFlag & DEBUG_WARN) { taosPrintLog("MND WARN ", 255, __VA_ARGS__); }}
#define mInfo(...) { if (mDebugFlag & DEBUG_INFO) { taosPrintLog("MND ", 255, __VA_ARGS__); }}
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); }}
#define mFatal(...) { if (mDebugFlag & DEBUG_FATAL) { taosPrintLog("MND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}
#define mError(...) { if (mDebugFlag & DEBUG_ERROR) { taosPrintLog("MND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}
#define mWarn(...) { if (mDebugFlag & DEBUG_WARN) { taosPrintLog("MND WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}
#define mInfo(...) { if (mDebugFlag & DEBUG_INFO) { taosPrintLog("MND ", DEBUG_INFO, 255, __VA_ARGS__); }}
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", DEBUG_DEBUG, mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", DEBUG_TRACE, mDebugFlag, __VA_ARGS__); }}
#define SDB_MAX_SIZE (32 * 1024)
......
source/dnode/vnode/inc/meta.h
浏览文件 @ 78513e2e
......@@ -24,8 +24,8 @@
extern "C" {
#endif
#define META_SUPER_TABLE TD_SUPER_TABLE
#define META_CHILD_TABLE TD_CHILD_TABLE
#define META_SUPER_TABLE TD_SUPER_TABLE
#define META_CHILD_TABLE TD_CHILD_TABLE
#define META_NORMAL_TABLE TD_NORMAL_TABLE
// Types exported
......@@ -50,14 +50,14 @@ int metaDropTable(SMeta *pMeta, tb_uid_t uid);
int metaCommit(SMeta *pMeta);
// For Query
STbCfg * metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid);
STbCfg * metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid);
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid);
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid);
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline);
STSchema * metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver);
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver);
SMTbCursor *metaOpenTbCursor(SMeta *pMeta);
void metaCloseTbCursor(SMTbCursor *pTbCur);
char * metaTbCursorNext(SMTbCursor *pTbCur);
char *metaTbCursorNext(SMTbCursor *pTbCur);
SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid);
void metaCloseCtbCurosr(SMCtbCursor *pCtbCur);
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 78513e2e
......@@ -57,6 +57,9 @@ typedef struct {
SMetaCfg metaCfg;
STqCfg tqCfg;
SWalCfg walCfg;
uint32_t hashBegin;
uint32_t hashEnd;
int8_t hashMethod;
} SVnodeCfg;
typedef struct {
......@@ -210,6 +213,10 @@ static FORCE_INLINE void tqReadHandleSetColIdList(STqReadHandle *pReadHandle, SA
//}
static FORCE_INLINE int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList) {
if (pHandle->tbIdHash) {
taosHashClear(pHandle->tbIdHash);
}
pHandle->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
if (pHandle->tbIdHash == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
......@@ -224,6 +231,23 @@ static FORCE_INLINE int tqReadHandleSetTbUidList(STqReadHandle *pHandle, const S
return 0;
}
static FORCE_INLINE int tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList) {
if (pHandle->tbIdHash == NULL) {
pHandle->tbIdHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
if (pHandle->tbIdHash == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
for (int i = 0; i < taosArrayGetSize(tbUidList); i++) {
int64_t *pKey = (int64_t *)taosArrayGet(tbUidList, i);
taosHashPut(pHandle->tbIdHash, pKey, sizeof(int64_t), NULL, 0);
}
return 0;
}
int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle);
int tqRetrieveDataBlockInfo(STqReadHandle *pHandle, SDataBlockInfo *pBlockInfo);
......
source/dnode/vnode/src/inc/metaDef.h
浏览文件 @ 78513e2e
......@@ -76,4 +76,4 @@ struct SMeta {
}
#endif
#endif /*_TD_META_DEF_H_*/
\ No newline at end of file
#endif /*_TD_META_DEF_H_*/
source/dnode/vnode/src/inc/tqInt.h
浏览文件 @ 78513e2e
......@@ -25,43 +25,41 @@
extern "C" {
#endif
extern int32_t tqDebugFlag;
#define tqFatal(...) \
{ \
if (tqDebugFlag & DEBUG_FATAL) { \
taosPrintLog("TQ FATAL ", 255, __VA_ARGS__); \
} \
#define tqFatal(...) \
{ \
if (tqDebugFlag & DEBUG_FATAL) { \
taosPrintLog("TQ FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \
} \
}
#define tqError(...) \
{ \
if (tqDebugFlag & DEBUG_ERROR) { \
taosPrintLog("TQ ERROR ", 255, __VA_ARGS__); \
} \
#define tqError(...) \
{ \
if (tqDebugFlag & DEBUG_ERROR) { \
taosPrintLog("TQ ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \
} \
}
#define tqWarn(...) \
{ \
if (tqDebugFlag & DEBUG_WARN) { \
taosPrintLog("TQ WARN ", 255, __VA_ARGS__); \
} \
#define tqWarn(...) \
{ \
if (tqDebugFlag & DEBUG_WARN) { \
taosPrintLog("TQ WARN ", DEBUG_WARN, 255, __VA_ARGS__); \
} \
}
#define tqInfo(...) \
{ \
if (tqDebugFlag & DEBUG_INFO) { \
taosPrintLog("TQ ", 255, __VA_ARGS__); \
} \
#define tqInfo(...) \
{ \
if (tqDebugFlag & DEBUG_INFO) { \
taosPrintLog("TQ ", DEBUG_INFO, 255, __VA_ARGS__); \
} \
}
#define tqDebug(...) \
{ \
if (tqDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("TQ ", tqDebugFlag, __VA_ARGS__); \
} \
#define tqDebug(...) \
{ \
if (tqDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("TQ ", DEBUG_DEBUG, tqDebugFlag, __VA_ARGS__); \
} \
}
#define tqTrace(...) \
{ \
if (tqDebugFlag & DEBUG_TRACE) { \
taosPrintLog("TQ ", tqDebugFlag, __VA_ARGS__); \
} \
#define tqTrace(...) \
{ \
if (tqDebugFlag & DEBUG_TRACE) { \
taosPrintLog("TQ ", DEBUG_TRACE, tqDebugFlag, __VA_ARGS__); \
} \
}
#define TQ_BUFFER_SIZE 8
......
source/dnode/vnode/src/inc/tsdbLog.h
浏览文件 @ 78513e2e
......@@ -24,12 +24,12 @@ extern "C" {
extern int32_t tsdbDebugFlag;
#define tsdbFatal(...) do { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", 255, __VA_ARGS__); }} while(0)
#define tsdbError(...) do { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", 255, __VA_ARGS__); }} while(0)
#define tsdbWarn(...) do { if (tsdbDebugFlag & DEBUG_WARN) { taosPrintLog("TDB WARN ", 255, __VA_ARGS__); }} while(0)
#define tsdbInfo(...) do { if (tsdbDebugFlag & DEBUG_INFO) { taosPrintLog("TDB ", 255, __VA_ARGS__); }} while(0)
#define tsdbDebug(...) do { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", tsdbDebugFlag, __VA_ARGS__); }} while(0)
#define tsdbFatal(...) do { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TDB FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }} while(0)
#define tsdbError(...) do { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TDB ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }} while(0)
#define tsdbWarn(...) do { if (tsdbDebugFlag & DEBUG_WARN) { taosPrintLog("TDB WARN ", DEBUG_WARN, 255, __VA_ARGS__); }} while(0)
#define tsdbInfo(...) do { if (tsdbDebugFlag & DEBUG_INFO) { taosPrintLog("TDB ", DEBUG_INFO, 255, __VA_ARGS__); }} while(0)
#define tsdbDebug(...) do { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TDB ", DEBUG_DEBUG, tsdbDebugFlag, __VA_ARGS__); }} while(0)
#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TDB ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
#ifdef __cplusplus
}
......
source/dnode/vnode/src/inc/vnd.h
浏览文件 @ 78513e2e
......@@ -88,44 +88,41 @@ int vnodeScheduleTask(SVnodeTask* task);
int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq);
void vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq);
// For Log
extern int32_t vDebugFlag;
#define vFatal(...) \
do { \
if (vDebugFlag & DEBUG_FATAL) { \
taosPrintLog("VND FATAL ", 255, __VA_ARGS__); \
} \
#define vFatal(...) \
do { \
if (vDebugFlag & DEBUG_FATAL) { \
taosPrintLog("VND FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); \
} \
} while (0)
#define vError(...) \
do { \
if (vDebugFlag & DEBUG_ERROR) { \
taosPrintLog("VND ERROR ", 255, __VA_ARGS__); \
} \
#define vError(...) \
do { \
if (vDebugFlag & DEBUG_ERROR) { \
taosPrintLog("VND ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); \
} \
} while (0)
#define vWarn(...) \
do { \
if (vDebugFlag & DEBUG_WARN) { \
taosPrintLog("VND WARN ", 255, __VA_ARGS__); \
} \
#define vWarn(...) \
do { \
if (vDebugFlag & DEBUG_WARN) { \
taosPrintLog("VND WARN ", DEBUG_WARN, 255, __VA_ARGS__); \
} \
} while (0)
#define vInfo(...) \
do { \
if (vDebugFlag & DEBUG_INFO) { \
taosPrintLog("VND ", 255, __VA_ARGS__); \
} \
#define vInfo(...) \
do { \
if (vDebugFlag & DEBUG_INFO) { \
taosPrintLog("VND ", DEBUG_INFO, 255, __VA_ARGS__); \
} \
} while (0)
#define vDebug(...) \
do { \
if (vDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("VND ", tsdbDebugFlag, __VA_ARGS__); \
} \
#define vDebug(...) \
do { \
if (vDebugFlag & DEBUG_DEBUG) { \
taosPrintLog("VND ", DEBUG_DEBUG, tsdbDebugFlag, __VA_ARGS__); \
} \
} while (0)
#define vTrace(...) \
do { \
if (vDebugFlag & DEBUG_TRACE) { \
taosPrintLog("VND ", tsdbDebugFlag, __VA_ARGS__); \
} \
#define vTrace(...) \
do { \
if (vDebugFlag & DEBUG_TRACE) { \
taosPrintLog("VND ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); \
} \
} while (0)
// vnodeCfg.h
......
source/dnode/vnode/src/meta/metaBDBImpl.c
浏览文件 @ 78513e2e
......@@ -64,10 +64,10 @@ static int metaStbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *
static int metaNtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
static int metaCtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey);
static int metaEncodeTbInfo(void **buf, STbCfg *pTbCfg);
static void * metaDecodeTbInfo(void *buf, STbCfg *pTbCfg);
static void *metaDecodeTbInfo(void *buf, STbCfg *pTbCfg);
static void metaClearTbCfg(STbCfg *pTbCfg);
static int metaEncodeSchema(void **buf, SSchemaWrapper *pSW);
static void * metaDecodeSchema(void *buf, SSchemaWrapper *pSW);
static void *metaDecodeSchema(void *buf, SSchemaWrapper *pSW);
static void metaDBWLock(SMetaDB *pDB);
static void metaDBRLock(SMetaDB *pDB);
static void metaDBULock(SMetaDB *pDB);
......@@ -150,7 +150,7 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
tb_uid_t uid;
char buf[512];
char buf1[512];
void * pBuf;
void *pBuf;
DBT key1, value1;
DBT key2, value2;
SSchema *pSchema = NULL;
......@@ -402,7 +402,7 @@ static int metaNtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey
static int metaCtbIdxCb(DB *pIdx, const DBT *pKey, const DBT *pValue, DBT *pSKey) {
STbCfg *pTbCfg = (STbCfg *)(pValue->app_data);
DBT * pDbt;
DBT *pDbt;
if (pTbCfg->type == META_CHILD_TABLE) {
// pDbt = calloc(2, sizeof(DBT));
......@@ -487,7 +487,7 @@ static void metaClearTbCfg(STbCfg *pTbCfg) {
/* ------------------------ FOR QUERY ------------------------ */
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
STbCfg * pTbCfg = NULL;
STbCfg *pTbCfg = NULL;
SMetaDB *pDB = pMeta->pDB;
DBT key = {0};
DBT value = {0};
......@@ -517,7 +517,7 @@ STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
}
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
STbCfg * pTbCfg = NULL;
STbCfg *pTbCfg = NULL;
SMetaDB *pDB = pMeta->pDB;
DBT key = {0};
DBT pkey = {0};
......@@ -551,10 +551,10 @@ STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
uint32_t nCols;
SSchemaWrapper *pSW = NULL;
SMetaDB * pDB = pMeta->pDB;
SMetaDB *pDB = pMeta->pDB;
int ret;
void * pBuf;
SSchema * pSchema;
void *pBuf;
SSchema *pSchema;
SSchemaKey schemaKey = {uid, sver, 0};
DBT key = {0};
DBT value = {0};
......@@ -586,7 +586,7 @@ struct SMTbCursor {
SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
SMTbCursor *pTbCur = NULL;
SMetaDB * pDB = pMeta->pDB;
SMetaDB *pDB = pMeta->pDB;
pTbCur = (SMTbCursor *)calloc(1, sizeof(*pTbCur));
if (pTbCur == NULL) {
......@@ -617,7 +617,7 @@ char *metaTbCursorNext(SMTbCursor *pTbCur) {
DBT key = {0};
DBT value = {0};
STbCfg tbCfg;
void * pBuf;
void *pBuf;
for (;;) {
if (pTbCur->pCur->get(pTbCur->pCur, &key, &value, DB_NEXT) == 0) {
......@@ -639,10 +639,10 @@ char *metaTbCursorNext(SMTbCursor *pTbCur) {
STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
STSchemaBuilder sb;
STSchema * pTSchema = NULL;
SSchema * pSchema;
STSchema *pTSchema = NULL;
SSchema *pSchema;
SSchemaWrapper *pSW;
STbCfg * pTbCfg;
STbCfg *pTbCfg;
tb_uid_t quid;
pTbCfg = metaGetTbInfoByUid(pMeta, uid);
......@@ -670,13 +670,13 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
}
struct SMCtbCursor {
DBC * pCur;
DBC *pCur;
tb_uid_t suid;
};
SMCtbCursor *metaOpenCtbCursor(SMeta *pMeta, tb_uid_t uid) {
SMCtbCursor *pCtbCur = NULL;
SMetaDB * pDB = pMeta->pDB;
SMetaDB *pDB = pMeta->pDB;
int ret;
pCtbCur = (SMCtbCursor *)calloc(1, sizeof(*pCtbCur));
......@@ -708,7 +708,7 @@ tb_uid_t metaCtbCursorNext(SMCtbCursor *pCtbCur) {
DBT skey = {0};
DBT pkey = {0};
DBT pval = {0};
void * pBuf;
void *pBuf;
STbCfg tbCfg;
// Set key
......
source/dnode/vnode/src/tq/tq.c
浏览文件 @ 78513e2e
......@@ -72,6 +72,8 @@ void tqClose(STQ* pTq) {
}
int tqPushMsg(STQ* pTq, void* msg, tmsg_t msgType, int64_t version) {
// if waiting
// memcpy and send msg to fetch thread
// TODO: add reference
// if handle waiting, launch query and response to consumer
//
......@@ -210,7 +212,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
SMqConsumeReq* pReq = pMsg->pCont;
int64_t consumerId = pReq->consumerId;
int64_t fetchOffset;
/*int64_t blockingTime = pReq->blockingTime;*/
int64_t blockingTime = pReq->blockingTime;
if (pReq->currentOffset == TMQ_CONF__RESET_OFFSET__EARLIEAST) {
fetchOffset = 0;
......
source/dnode/vnode/src/tq/tqMetaStore.c
浏览文件 @ 78513e2e
......@@ -90,7 +90,7 @@ STqMetaStore* tqStoreOpen(STQ* pTq, const char* path, FTqSerialize serializer, F
char name[pathLen + 10];
strcpy(name, path);
if (taosDirExist(name) != 0 && taosMkDir(name) != 0) {
if (!taosDirExist(name) && taosMkDir(name) != 0) {
terrno = TSDB_CODE_TQ_FAILED_TO_CREATE_DIR;
tqError("failed to create dir:%s since %s ", name, terrstr());
}
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ 78513e2e
......@@ -580,7 +580,7 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile *pRSmadF = TSDB_READ_SMAD_FILE(&(pCommith->readh));
SDFile *pWSmadF = TSDB_COMMIT_SMAD_FILE(pCommith);
if (access(TSDB_FILE_FULL_NAME(pRSmadF), F_OK) != 0) {
if (!taosCheckExistFile(TSDB_FILE_FULL_NAME(pRSmadF))) {
tsdbDebug("vgId:%d create data file %s as not exist", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pRSmadF));
tsdbInitDFile(pRepo, pWSmadF, did, fid, FS_TXN_VERSION(REPO_FS(pRepo)), TSDB_FILE_SMAD);
......@@ -614,7 +614,7 @@ static int tsdbSetAndOpenCommitFile(SCommitH *pCommith, SDFileSet *pSet, int fid
SDFile *pRSmalF = TSDB_READ_SMAL_FILE(&(pCommith->readh));
SDFile *pWSmalF = TSDB_COMMIT_SMAL_FILE(pCommith);
if ((pCommith->isLFileSame) && access(TSDB_FILE_FULL_NAME(pRSmalF), F_OK) == 0) {
if ((pCommith->isLFileSame) && taosCheckExistFile(TSDB_FILE_FULL_NAME(pRSmalF))) {
tsdbInitDFileEx(pWSmalF, pRSmalF);
if (tsdbOpenDFile(pWSmalF, O_RDWR) < 0) {
tsdbError("vgId:%d failed to open file %s to commit since %s", REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pWSmalF),
......@@ -1054,7 +1054,7 @@ int tsdbWriteBlockIdx(SDFile *pHeadf, SArray *pIdxA, void **ppBuf) {
// pfs->metaCacheComp = NULL;
// } else {
// // remove meta.tmp file
// remove(mf.f.aname);
// taosRemoveFile(mf.f.aname);
// taosHashCleanup(pfs->metaCacheComp);
// pfs->metaCacheComp = NULL;
// }
......
source/dnode/vnode/src/tsdb/tsdbFS.c
浏览文件 @ 78513e2e
......@@ -314,7 +314,7 @@ int tsdbOpenFS(STsdb *pRepo) {
tsdbGetTxnFname(pRepo, TSDB_TXN_CURR_FILE, current);
tsdbGetRtnSnap(pRepo, &pRepo->rtn);
if (access(current, F_OK) == 0) {
if (taosCheckExistFile(current)) {
if (tsdbOpenFSFromCurrent(pRepo) < 0) {
tsdbError("vgId:%d failed to open FS since %s", REPO_ID(pRepo), tstrerror(terrno));
return -1;
......@@ -439,7 +439,7 @@ static int tsdbSaveFSStatus(STsdb *pRepo, SFSStatus *pStatus) {
if (taosWriteFile(pFile, hbuf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCloseFile(&pFile);
remove(tfname);
taosRemoveFile(tfname);
return -1;
}
......@@ -447,7 +447,7 @@ static int tsdbSaveFSStatus(STsdb *pRepo, SFSStatus *pStatus) {
if (fsheader.len > 0) {
if (tsdbMakeRoom(&(pBuf), fsheader.len) < 0) {
taosCloseFile(&pFile);
remove(tfname);
taosRemoveFile(tfname);
return -1;
}
......@@ -458,7 +458,7 @@ static int tsdbSaveFSStatus(STsdb *pRepo, SFSStatus *pStatus) {
if (taosWriteFile(pFile, pBuf, fsheader.len) < fsheader.len) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCloseFile(&pFile);
(void)remove(tfname);
(void)taosRemoveFile(tfname);
taosTZfree(pBuf);
return -1;
}
......@@ -468,7 +468,7 @@ static int tsdbSaveFSStatus(STsdb *pRepo, SFSStatus *pStatus) {
if (taosFsyncFile(pFile) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
taosCloseFile(&pFile);
remove(tfname);
taosRemoveFile(tfname);
taosTZfree(pBuf);
return -1;
}
......
source/dnode/vnode/src/tsdb/tsdbFile.c
浏览文件 @ 78513e2e
......@@ -443,25 +443,24 @@ int tsdbLoadDFileHeader(SDFile *pDFile, SDFInfo *pInfo) {
}
static int tsdbScanAndTryFixDFile(STsdb *pRepo, SDFile *pDFile) {
struct stat dfstat;
SDFile df;
tsdbInitDFileEx(&df, pDFile);
if (access(TSDB_FILE_FULL_NAME(pDFile), F_OK) != 0) {
if (!taosCheckExistFile(TSDB_FILE_FULL_NAME(pDFile))) {
tsdbError("vgId:%d data file %s not exit, report to upper layer to fix it", REPO_ID(pRepo),
TSDB_FILE_FULL_NAME(pDFile));
// pRepo->state |= TSDB_STATE_BAD_DATA;
TSDB_FILE_SET_STATE(pDFile, TSDB_FILE_STATE_BAD);
return 0;
}
if (stat(TSDB_FILE_FULL_NAME(&df), &dfstat) < 0) {
int64_t file_size = 0;
if (taosStatFile(TSDB_FILE_FULL_NAME(&df), &file_size, NULL) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (pDFile->info.size < dfstat.st_size) {
if (pDFile->info.size < file_size) {
// if (tsdbOpenDFile(&df, O_WRONLY) < 0) {
if (tsdbOpenDFile(&df, TD_FILE_WRITE) < 0) {
return -1;
......@@ -480,10 +479,10 @@ static int tsdbScanAndTryFixDFile(STsdb *pRepo, SDFile *pDFile) {
tsdbCloseDFile(&df);
tsdbInfo("vgId:%d file %s is truncated from %" PRId64 " to %" PRId64, REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile),
dfstat.st_size, pDFile->info.size);
} else if (pDFile->info.size > dfstat.st_size) {
file_size, pDFile->info.size);
} else if (pDFile->info.size > file_size) {
tsdbError("vgId:%d data file %s has wrong size %" PRId64 " expected %" PRId64 ", report to upper layer to fix it",
REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), dfstat.st_size, pDFile->info.size);
REPO_ID(pRepo), TSDB_FILE_FULL_NAME(pDFile), file_size, pDFile->info.size);
// pRepo->state |= TSDB_STATE_BAD_DATA;
TSDB_FILE_SET_STATE(pDFile, TSDB_FILE_STATE_BAD);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
......
source/dnode/vnode/src/vnd/vnodeMain.c
浏览文件 @ 78513e2e
......@@ -30,6 +30,9 @@ SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
cfg.pDnode = pVnodeCfg->pDnode;
cfg.pTfs = pVnodeCfg->pTfs;
cfg.dbId = pVnodeCfg->dbId;
cfg.hashBegin = pVnodeCfg->hashBegin;
cfg.hashEnd = pVnodeCfg->hashEnd;
cfg.hashMethod = pVnodeCfg->hashMethod;
}
// Validate options
......
source/libs/CMakeLists.txt
浏览文件 @ 78513e2e
add_definitions("-D ALLOW_FORBID_FUNC")
add_subdirectory(transport)
add_subdirectory(sync)
add_subdirectory(tdb)
......
source/libs/catalog/inc/catalogInt.h
浏览文件 @ 78513e2e
......@@ -60,6 +60,7 @@ typedef struct SCtgDebug {
bool lockDebug;
bool cacheDebug;
bool apiDebug;
bool metaDebug;
uint32_t showCachePeriodSec;
} SCtgDebug;
......@@ -119,6 +120,10 @@ typedef struct SCatalogStat {
SCtgCacheStat cache;
} SCatalogStat;
typedef struct SCtgUpdateMsgHeader {
SCatalog* pCtg;
} SCtgUpdateMsgHeader;
typedef struct SCtgUpdateVgMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
......@@ -145,6 +150,14 @@ typedef struct SCtgRemoveStbMsg {
uint64_t suid;
} SCtgRemoveStbMsg;
typedef struct SCtgRemoveTblMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
char tbName[TSDB_TABLE_NAME_LEN];
uint64_t dbId;
} SCtgRemoveTblMsg;
typedef struct SCtgMetaAction {
int32_t act;
void *data;
......@@ -189,19 +202,21 @@ typedef struct SCtgAction {
#define CTG_IS_META_TABLE(type) ((type) == META_TYPE_TABLE)
#define CTG_IS_META_BOTH(type) ((type) == META_TYPE_BOTH_TABLE)
#define CTG_FLAG_STB 0x1
#define CTG_FLAG_NOT_STB 0x2
#define CTG_FLAG_UNKNOWN_STB 0x4
#define CTG_FLAG_INF_DB 0x8
#define CTG_IS_STB(_flag) ((_flag) & CTG_FLAG_STB)
#define CTG_IS_NOT_STB(_flag) ((_flag) & CTG_FLAG_NOT_STB)
#define CTG_IS_UNKNOWN_STB(_flag) ((_flag) & CTG_FLAG_UNKNOWN_STB)
#define CTG_IS_INF_DB(_flag) ((_flag) & CTG_FLAG_INF_DB)
#define CTG_SET_INF_DB(_flag) ((_flag) |= CTG_FLAG_INF_DB)
#define CTG_SET_STB(_flag, tbType) do { (_flag) |= ((tbType) == TSDB_SUPER_TABLE) ? CTG_FLAG_STB : ((tbType) > TSDB_SUPER_TABLE ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB); } while (0)
#define CTG_GEN_STB_FLAG(_isStb) ((_isStb) == 1) ? CTG_FLAG_STB : ((_isStb) == 0 ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB)
#define CTG_TBTYPE_MATCH(_flag, tbType) (CTG_IS_UNKNOWN_STB(_flag) || (CTG_IS_STB(_flag) && (tbType) == TSDB_SUPER_TABLE) || (CTG_IS_NOT_STB(_flag) && (tbType) != TSDB_SUPER_TABLE))
#define CTG_FLAG_STB 0x1
#define CTG_FLAG_NOT_STB 0x2
#define CTG_FLAG_UNKNOWN_STB 0x4
#define CTG_FLAG_INF_DB 0x8
#define CTG_FLAG_FORCE_UPDATE 0x10
#define CTG_FLAG_IS_STB(_flag) ((_flag) & CTG_FLAG_STB)
#define CTG_FLAG_IS_NOT_STB(_flag) ((_flag) & CTG_FLAG_NOT_STB)
#define CTG_FLAG_IS_UNKNOWN_STB(_flag) ((_flag) & CTG_FLAG_UNKNOWN_STB)
#define CTG_FLAG_IS_INF_DB(_flag) ((_flag) & CTG_FLAG_INF_DB)
#define CTG_FLAG_IS_FORCE_UPDATE(_flag) ((_flag) & CTG_FLAG_FORCE_UPDATE)
#define CTG_FLAG_SET_INF_DB(_flag) ((_flag) |= CTG_FLAG_INF_DB)
#define CTG_FLAG_SET_STB(_flag, tbType) do { (_flag) |= ((tbType) == TSDB_SUPER_TABLE) ? CTG_FLAG_STB : ((tbType) > TSDB_SUPER_TABLE ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB); } while (0)
#define CTG_FLAG_MAKE_STB(_isStb) (((_isStb) == 1) ? CTG_FLAG_STB : ((_isStb) == 0 ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB))
#define CTG_FLAG_MATCH_STB(_flag, tbType) (CTG_FLAG_IS_UNKNOWN_STB(_flag) || (CTG_FLAG_IS_STB(_flag) && (tbType) == TSDB_SUPER_TABLE) || (CTG_FLAG_IS_NOT_STB(_flag) && (tbType) != TSDB_SUPER_TABLE))
#define CTG_IS_INF_DBNAME(_dbname) ((*(_dbname) == 'i') && (0 == strcmp(_dbname, TSDB_INFORMATION_SCHEMA_DB)))
......
source/libs/catalog/src/catalog.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/executor/inc/executil.h
浏览文件 @ 78513e2e
......@@ -68,7 +68,7 @@ typedef struct SResultRow {
} SResultRow;
typedef struct SResultRowInfo {
SResultRow *pCurResult; // current active result row info
SList* pRows;
SResultRow** pResult; // result list
// int16_t type:8; // data type for hash key
int32_t size; // number of result set
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ 78513e2e
......@@ -15,12 +15,12 @@
#ifndef TDENGINE_EXECUTORIMPL_H
#define TDENGINE_EXECUTORIMPL_H
#include "tsort.h"
#ifdef __cplusplus
extern "C" {
#endif
#include "os.h"
#include "tsort.h"
#include "tcommon.h"
#include "tlosertree.h"
#include "ttszip.h"
......@@ -157,6 +157,13 @@ typedef struct STaskCostInfo {
SHashObj* operatorProfResults; // map<operator_type, SQueryProfEvent>
} STaskCostInfo;
typedef struct SOperatorCostInfo {
uint64_t openCost;
uint64_t execCost;
uint64_t totalRows;
uint64_t totalBytes;
} SOperatorCostInfo;
typedef struct {
int64_t vgroupLimit;
int64_t ts;
......@@ -233,9 +240,9 @@ typedef struct STaskAttr {
SArray* pUdfInfo; // no need to free
} STaskAttr;
typedef int32_t (*__optr_prepare_fn_t)(void* param);
typedef SSDataBlock* (*__operator_fn_t)(void* param, bool* newgroup);
typedef void (*__optr_cleanup_fn_t)(void* param, int32_t num);
typedef int32_t (*__optr_open_fn_t)(void* param);
typedef SSDataBlock* (*__optr_fn_t)(void* param, bool* newgroup);
typedef void (*__optr_close_fn_t)(void* param, int32_t num);
struct SOperatorInfo;
......@@ -306,21 +313,21 @@ enum {
};
typedef struct SOperatorInfo {
uint8_t operatorType;
bool blockingOptr; // block operator or not
uint8_t status; // denote if current operator is completed
int32_t numOfOutput; // number of columns of the current operator results
char* name; // name, used to show the query execution plan
void* info; // extension attribution
SExprInfo* pExpr;
STaskRuntimeEnv* pRuntimeEnv; // todo remove it
SExecTaskInfo* pTaskInfo;
uint8_t operatorType;
bool blockingOptr; // block operator or not
uint8_t status; // denote if current operator is completed
int32_t numOfOutput; // number of columns of the current operator results
char* name; // name, used to show the query execution plan
void* info; // extension attribution
SExprInfo* pExpr;
STaskRuntimeEnv* pRuntimeEnv; // todo remove it
SExecTaskInfo* pTaskInfo;
struct SOperatorInfo** pDownstream; // downstram pointer list
int32_t numOfDownstream; // number of downstream. The value is always ONE expect for join operator
__optr_prepare_fn_t prepareFn;
__operator_fn_t exec;
__optr_cleanup_fn_t cleanupFn;
__optr_open_fn_t openFn;
__optr_fn_t nextDataFn;
__optr_close_fn_t closeFn;
} SOperatorInfo;
typedef struct {
......@@ -479,9 +486,6 @@ typedef struct SAggOperatorInfo {
typedef struct SProjectOperatorInfo {
SOptrBasicInfo binfo;
int32_t bufCapacity;
uint32_t seed;
SSDataBlock* existDataBlock;
} SProjectOperatorInfo;
......@@ -615,10 +619,10 @@ SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order,
SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
SOperatorInfo* createProjectOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream, SExprInfo* pExpr,
int32_t numOfOutput);
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream);
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SInterval* pInterval, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createAllTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
SExprInfo* pExpr, int32_t numOfOutput);
......@@ -654,8 +658,6 @@ SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pdownstream, int32_t numOf
SOperatorInfo* createOrderOperatorInfo(SOperatorInfo* downstream, SArray* pExprInfo, SArray* pOrderVal, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SArray* pExprInfo, SArray* pOrderVal, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
// SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
// SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
// SSDataBlock* doSLimit(void* param, bool* newgroup);
// int32_t doCreateFilterInfo(SColumnInfo* pCols, int32_t numOfCols, int32_t numOfFilterCols, SSingleColumnFilterInfo** pFilterInfo, uint64_t qId);
......
source/libs/executor/inc/tsimplehash.h 0 → 100644
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/executor/src/executor.c
浏览文件 @ 78513e2e
......@@ -95,17 +95,17 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle) {
}
int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, SArray* tableIdList, bool isAdd) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo* )tinfo;
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
// traverse to the streamscan node to add this table id
SOperatorInfo* pInfo = pTaskInfo->pRoot;
while(pInfo->operatorType != OP_StreamScan) {
while (pInfo->operatorType != OP_StreamScan) {
pInfo = pInfo->pDownstream[0];
}
SStreamBlockScanInfo* pScanInfo = pInfo->info;
if (isAdd) {
int32_t code = tqReadHandleSetTbUidList(pScanInfo->readerHandle, tableIdList);
int32_t code = tqReadHandleAddTbUidList(pScanInfo->readerHandle, tableIdList);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
......@@ -114,4 +114,4 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, SArray* tableIdList, bool isA
}
return TSDB_CODE_SUCCESS;
}
\ No newline at end of file
}
source/libs/executor/src/tsimplehash.c 0 → 100644
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/executor/src/tsort.c
浏览文件 @ 78513e2e
......@@ -16,11 +16,11 @@
#include "tcommon.h"
#include "query.h"
#include "tsort.h"
#include "tep.h"
#include "tdatablock.h"
#include "tdef.h"
#include "tlosertree.h"
#include "tpagedbuf.h"
#include "tsort.h"
#include "tutil.h"
typedef struct STupleHandle {
......
source/libs/function/src/taggfunction.c
浏览文件 @ 78513e2e
......@@ -28,8 +28,8 @@
#include "tbuffer.h"
#include "tcompression.h"
//#include "queryLog.h"
#include "tdatablock.h"
#include "tudf.h"
#include "tep.h"
#define GET_INPUT_DATA_LIST(x) ((char *)((x)->pInput))
#define GET_INPUT_DATA(x, y) ((char*) colDataGetData((x)->pInput, (y)))
......
source/libs/index/inc/indexInt.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/inc/index_comm.h 0 → 100644
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/inc/index_util.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/src/index.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/src/index_cache.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/src/index_comm.c 0 → 100644
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/src/index_json.c 0 → 100644
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/src/index_tfile.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/src/index_util.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/test/fstTest.cc
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/test/fstUT.cc
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/test/jsonUT.cc 0 → 100644
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/index/test/utilUT.cc
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/monitor/inc/monInt.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/monitor/src/monitor.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/qcom/src/queryUtil.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/scalar/inc/filterInt.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/scalar/src/filter.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/scalar/src/scalar.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/scalar/src/sclvector.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/inc/syncInt.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/inc/syncMessage.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/inc/syncVoteMgr.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/src/syncIO.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/src/syncMain.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/src/syncMessage.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/sync/test/syncTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tdb/src/db/tdb.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tdb/src/db/tdbUtil.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tdb/src/inc/tdbUtil.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tdb/test/tdbTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tfs/CMakeLists.txt
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tfs/inc/tfsInt.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tfs/src/tfs.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/tfs/test/tfsTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/transport/inc/rpcLog.h
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/wal/src/walMeta.c
浏览文件 @ 78513e2e
此差异已折叠。
source/libs/wal/src/walWrite.c
浏览文件 @ 78513e2e
此差异已折叠。
source/os/src/osDir.c
浏览文件 @ 78513e2e
此差异已折叠。
source/os/src/osEnv.c
浏览文件 @ 78513e2e
此差异已折叠。
source/os/src/osFile.c
浏览文件 @ 78513e2e
此差异已折叠。
source/os/src/osSysinfo.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/tcache.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/tconfig.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/terror.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/thash.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/thttp.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/tlog.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/tpagedbuf.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/src/ttimer.c
浏览文件 @ 78513e2e
此差异已折叠。
source/util/test/arrayTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/util/test/cacheTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/util/test/encodeTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/util/test/freelistTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/util/test/hashTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
source/util/test/pageBufferTest.cpp
浏览文件 @ 78513e2e
此差异已折叠。
tests @ 904e6f0e
比较 08ed39f0...904e6f0e
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或