Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 73a3f9bb 编写于 作者: S Shengliang Guan
浏览文件
操作

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

上级 6e753618 631c9668
展开全部 隐藏空白更改
内联 并排
Showing with 3003 addition and 1645 deletion
examples/lua/build.sh
浏览文件 @ 73a3f9bb
......@@ -4,5 +4,5 @@ if [ "$lua_header_installed" = "0" ]; then
sudo apt install -y liblua5.3-dev
fi
gcc -std=c99 lua_connector.c -fPIC -shared -o luaconnector.so -Wall -ltaos -I/usr/include/lua5.3
gcc -std=c99 lua_connector.c -fPIC -shared -o luaconnector.so -Wall -ltaos -I/usr/include/lua5.3 -I../../include/client
examples/lua/lua_connector.c
浏览文件 @ 73a3f9bb
#include <lauxlib.h>
#include <lua.h>
#include <lualib.h>
#include <stdio.h>
#include <math.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include "../../../include/client/taos.h"
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>
#include "taos.h"
struct cb_param{
lua_State* state;
......@@ -28,14 +28,14 @@ static int l_connect(lua_State *L){
luaL_checktype(L, 1, LUA_TTABLE);
lua_getfield(L,1,"host");
lua_getfield(L, 1,"host");
if (lua_isstring(L,-1)){
host = lua_tostring(L, -1);
// printf("host = %s\n", host);
}
lua_getfield(L, 1, "port");
if (lua_isinteger(L,-1)){
if (lua_isinteger(L, -1)){
port = lua_tointeger(L, -1);
//printf("port = %d\n", port);
}
......@@ -60,8 +60,6 @@ static int l_connect(lua_State *L){
lua_settop(L,0);
taos_init();
lua_newtable(L);
int table_index = lua_gettop(L);
......@@ -102,7 +100,7 @@ static int l_query(lua_State *L){
printf("failed, reason:%s\n", taos_errstr(result));
lua_pushinteger(L, -1);
lua_setfield(L, table_index, "code");
lua_pushstring(L, taos_errstr(taos));
lua_pushstring(L, taos_errstr(result));
lua_setfield(L, table_index, "error");
return 1;
......@@ -113,7 +111,6 @@ static int l_query(lua_State *L){
int rows = 0;
int num_fields = taos_field_count(result);
const TAOS_FIELD *fields = taos_fetch_fields(result);
//char temp[256];
const int affectRows = taos_affected_rows(result);
// printf(" affect rows:%d\r\n", affectRows);
......@@ -122,7 +119,7 @@ static int l_query(lua_State *L){
lua_pushinteger(L, affectRows);
lua_setfield(L, table_index, "affected");
lua_newtable(L);
while ((row = taos_fetch_row(result))) {
//printf("row index:%d\n",rows);
rows++;
......@@ -136,17 +133,21 @@ static int l_query(lua_State *L){
}
lua_pushstring(L,fields[i].name);
int32_t* length = taos_fetch_lengths(result);
switch (fields[i].type) {
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_TINYINT:
lua_pushinteger(L,*((char *)row[i]));
break;
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_SMALLINT:
lua_pushinteger(L,*((short *)row[i]));
break;
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_INT:
lua_pushinteger(L,*((int *)row[i]));
break;
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_BIGINT:
lua_pushinteger(L,*((int64_t *)row[i]));
break;
......@@ -156,9 +157,11 @@ static int l_query(lua_State *L){
case TSDB_DATA_TYPE_DOUBLE:
lua_pushnumber(L,*((double *)row[i]));
break;
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
lua_pushstring(L,(char *)row[i]);
//printf("type:%d, max len:%d, current len:%d\n",fields[i].type, fields[i].bytes, length[i]);
lua_pushlstring(L,(char *)row[i], length[i]);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
lua_pushinteger(L,*((int64_t *)row[i]));
......@@ -166,6 +169,7 @@ static int l_query(lua_State *L){
case TSDB_DATA_TYPE_BOOL:
lua_pushinteger(L,*((char *)row[i]));
break;
case TSDB_DATA_TYPE_NULL:
default:
lua_pushnil(L);
break;
......@@ -235,112 +239,6 @@ static int l_async_query(lua_State *L){
return 1;
}
void stream_cb(void *param, TAOS_RES *result, TAOS_ROW row){
struct cb_param* p = (struct cb_param*) param;
TAOS_FIELD *fields = taos_fetch_fields(result);
int numFields = taos_num_fields(result);
// printf("\nnumfields:%d\n", numFields);
//printf("\n\r-----------------------------------------------------------------------------------\n");
lua_State *L = p->state;
lua_rawgeti(L, LUA_REGISTRYINDEX, p->callback);
lua_newtable(L);
for (int i = 0; i < numFields; ++i) {
if (row[i] == NULL) {
continue;
}
lua_pushstring(L,fields[i].name);
switch (fields[i].type) {
case TSDB_DATA_TYPE_TINYINT:
lua_pushinteger(L,*((char *)row[i]));
break;
case TSDB_DATA_TYPE_SMALLINT:
lua_pushinteger(L,*((short *)row[i]));
break;
case TSDB_DATA_TYPE_INT:
lua_pushinteger(L,*((int *)row[i]));
break;
case TSDB_DATA_TYPE_BIGINT:
lua_pushinteger(L,*((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_FLOAT:
lua_pushnumber(L,*((float *)row[i]));
break;
case TSDB_DATA_TYPE_DOUBLE:
lua_pushnumber(L,*((double *)row[i]));
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
lua_pushstring(L,(char *)row[i]);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
lua_pushinteger(L,*((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_BOOL:
lua_pushinteger(L,*((char *)row[i]));
break;
default:
lua_pushnil(L);
break;
}
lua_settable(L, -3);
}
lua_call(L, 1, 0);
// printf("-----------------------------------------------------------------------------------\n\r");
}
static int l_open_stream(lua_State *L){
int r = luaL_ref(L, LUA_REGISTRYINDEX);
TAOS * taos = (TAOS*)lua_topointer(L,1);
const char * sqlstr = lua_tostring(L,2);
int stime = luaL_checknumber(L,3);
lua_newtable(L);
int table_index = lua_gettop(L);
struct cb_param *p = malloc(sizeof(struct cb_param));
p->state = L;
p->callback=r;
// printf("r:%d, L:%d\n",r,L);
void * s = taos_open_stream(taos,sqlstr,stream_cb,stime,p,NULL);
if (s == NULL) {
printf("failed to open stream, reason:%s\n", taos_errstr(taos));
free(p);
lua_pushnumber(L, -1);
lua_setfield(L, table_index, "code");
lua_pushstring(L, taos_errstr(taos));
lua_setfield(L, table_index, "error");
lua_pushlightuserdata(L,NULL);
lua_setfield(L, table_index, "stream");
}else{
// printf("success to open stream\n");
lua_pushnumber(L, 0);
lua_setfield(L, table_index, "code");
lua_pushstring(L, taos_errstr(taos));
lua_setfield(L, table_index, "error");
p->stream = s;
lua_pushlightuserdata(L,p);
lua_setfield(L, table_index, "stream");//stream has different content in lua and c.
}
return 1;
}
static int l_close_stream(lua_State *L){
//TODO:get stream and free cb_param
struct cb_param *p = lua_touserdata(L,1);
taos_close_stream(p->stream);
free(p);
return 0;
}
static int l_close(lua_State *L){
TAOS *taos= (TAOS*)lua_topointer(L,1);
......@@ -367,8 +265,6 @@ static const struct luaL_Reg lib[] = {
{"query", l_query},
{"query_a",l_async_query},
{"close", l_close},
{"open_stream", l_open_stream},
{"close_stream", l_close_stream},
{NULL, NULL}
};
......
include/common/tmsg.h
浏览文件 @ 73a3f9bb
......@@ -70,13 +70,11 @@ typedef uint16_t tmsg_t;
#define TSDB_IE_TYPE_DNODE_EXT 6
#define TSDB_IE_TYPE_DNODE_STATE 7
typedef enum {
HEARTBEAT_TYPE_MQ = 0,
HEARTBEAT_TYPE_QUERY,
// types can be added here
//
HEARTBEAT_TYPE_MAX
} EHbType;
enum {
CONN_TYPE__QUERY = 1,
CONN_TYPE__TMQ,
CONN_TYPE__MAX
};
enum {
HEARTBEAT_KEY_DBINFO = 1,
......@@ -346,7 +344,7 @@ int32_t tDeserializeSConnectReq(void* buf, int32_t bufLen, SConnectReq* pReq);
typedef struct {
int32_t acctId;
int64_t clusterId;
int32_t connId;
uint32_t connId;
int8_t superUser;
int8_t connType;
SEpSet epSet;
......@@ -1048,40 +1046,6 @@ typedef struct {
int32_t tSerializeSDCreateMnodeReq(void* buf, int32_t bufLen, SDCreateMnodeReq* pReq);
int32_t tDeserializeSDCreateMnodeReq(void* buf, int32_t bufLen, SDCreateMnodeReq* pReq);
typedef struct {
char sql[TSDB_SHOW_SQL_LEN];
int32_t queryId;
int64_t useconds;
int64_t stime;
int64_t qId;
int64_t sqlObjId;
int32_t pid;
char fqdn[TSDB_FQDN_LEN];
int8_t stableQuery;
int32_t numOfSub;
char subSqlInfo[TSDB_SHOW_SUBQUERY_LEN]; // include subqueries' index, Obj IDs and states(C-complete/I-imcomplete)
} SQueryDesc;
typedef struct {
int32_t connId;
int32_t pid;
int32_t numOfQueries;
int32_t numOfStreams;
char app[TSDB_APP_NAME_LEN];
char pData[];
} SHeartBeatReq;
typedef struct {
int32_t connId;
int32_t queryId;
int32_t streamId;
int32_t totalDnodes;
int32_t onlineDnodes;
int8_t killConnection;
int8_t align[3];
SEpSet epSet;
} SHeartBeatRsp;
typedef struct {
int32_t connId;
int32_t queryId;
......@@ -1684,13 +1648,48 @@ typedef struct {
} SKv;
typedef struct {
int32_t connId;
int32_t hbType;
int64_t tscRid;
int8_t connType;
} SClientHbKey;
typedef struct {
SClientHbKey connKey;
SHashObj* info; // hash<Skv.key, Skv>
int64_t tid;
int32_t status;
} SQuerySubDesc;
typedef struct {
char sql[TSDB_SHOW_SQL_LEN];
uint64_t queryId;
int64_t useconds;
int64_t stime;
int64_t reqRid;
int32_t pid;
char fqdn[TSDB_FQDN_LEN];
int32_t subPlanNum;
SArray* subDesc; // SArray<SQuerySubDesc>
} SQueryDesc;
typedef struct {
uint32_t connId;
int32_t pid;
char app[TSDB_APP_NAME_LEN];
SArray* queryDesc; // SArray<SQueryDesc>
} SQueryHbReqBasic;
typedef struct {
uint32_t connId;
uint64_t killRid;
int32_t totalDnodes;
int32_t onlineDnodes;
int8_t killConnection;
int8_t align[3];
SEpSet epSet;
} SQueryHbRspBasic;
typedef struct {
SClientHbKey connKey;
SQueryHbReqBasic* query;
SHashObj* info; // hash<Skv.key, Skv>
} SClientHbReq;
typedef struct {
......@@ -1699,9 +1698,10 @@ typedef struct {
} SClientHbBatchReq;
typedef struct {
SClientHbKey connKey;
int32_t status;
SArray* info; // Array<Skv>
SClientHbKey connKey;
int32_t status;
SQueryHbRspBasic* query;
SArray* info; // Array<Skv>
} SClientHbRsp;
typedef struct {
......@@ -1721,8 +1721,23 @@ static FORCE_INLINE void tFreeReqKvHash(SHashObj* info) {
}
}
static FORCE_INLINE void tFreeClientHbQueryDesc(void* pDesc) {
SQueryDesc* desc = (SQueryDesc*)pDesc;
if (desc->subDesc) {
taosArrayDestroy(desc->subDesc);
desc->subDesc = NULL;
}
}
static FORCE_INLINE void tFreeClientHbReq(void* pReq) {
SClientHbReq* req = (SClientHbReq*)pReq;
if (req->query) {
if (req->query->queryDesc) {
taosArrayDestroyEx(req->query->queryDesc, tFreeClientHbQueryDesc);
}
taosMemoryFreeClear(req->query);
}
if (req->info) {
tFreeReqKvHash(req->info);
taosHashCleanup(req->info);
......@@ -1751,6 +1766,7 @@ static FORCE_INLINE void tFreeClientKv(void* pKv) {
static FORCE_INLINE void tFreeClientHbRsp(void* pRsp) {
SClientHbRsp* rsp = (SClientHbRsp*)pRsp;
taosMemoryFreeClear(rsp->query);
if (rsp->info) taosArrayDestroyEx(rsp->info, tFreeClientKv);
}
......@@ -1779,14 +1795,14 @@ static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) {
}
static FORCE_INLINE int32_t tEncodeSClientHbKey(SCoder* pEncoder, const SClientHbKey* pKey) {
if (tEncodeI32(pEncoder, pKey->connId) < 0) return -1;
if (tEncodeI32(pEncoder, pKey->hbType) < 0) return -1;
if (tEncodeI64(pEncoder, pKey->tscRid) < 0) return -1;
if (tEncodeI8(pEncoder, pKey->connType) < 0) return -1;
return 0;
}
static FORCE_INLINE int32_t tDecodeSClientHbKey(SCoder* pDecoder, SClientHbKey* pKey) {
if (tDecodeI32(pDecoder, &pKey->connId) < 0) return -1;
if (tDecodeI32(pDecoder, &pKey->hbType) < 0) return -1;
if (tDecodeI64(pDecoder, &pKey->tscRid) < 0) return -1;
if (tDecodeI8(pDecoder, &pKey->connType) < 0) return -1;
return 0;
}
......
include/common/ttokendef.h
浏览文件 @ 73a3f9bb
......@@ -181,9 +181,9 @@
#define TK_NULL 163
#define TK_FIRST 164
#define TK_LAST 165
#define TK_CAST 166
#define TK_NOW 167
#define TK_TODAY 168
#define TK_NOW 166
#define TK_TODAY 167
#define TK_CAST 168
#define TK_ROWTS 169
#define TK_TBNAME 170
#define TK_QSTARTTS 171
......
include/libs/scalar/scalar.h
浏览文件 @ 73a3f9bb
......@@ -78,6 +78,8 @@ int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *
int32_t toUnixtimestampFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t nowFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t todayFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
bool getTimePseudoFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
......
include/libs/scheduler/scheduler.h
浏览文件 @ 73a3f9bb
......@@ -89,6 +89,8 @@ int32_t schedulerAsyncExecJob(void *transport, SArray *pNodeList, SQueryPlan* pD
*/
int32_t schedulerFetchRows(int64_t job, void **data);
int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub);
/**
* Cancel query job
......
include/os/osMemory.h
浏览文件 @ 73a3f9bb
......@@ -33,13 +33,13 @@ void *taosMemoryMalloc(int32_t size);
void *taosMemoryCalloc(int32_t num, int32_t size);
void *taosMemoryRealloc(void *ptr, int32_t size);
void *taosMemoryStrDup(void *ptr);
void taosMemoryFree(const void *ptr);
void taosMemoryFree(void *ptr);
int32_t taosMemorySize(void *ptr);
#define taosMemoryFreeClear(ptr) \
do { \
if (ptr) { \
taosMemoryFree(ptr); \
taosMemoryFree((void*)ptr); \
(ptr) = NULL; \
} \
} while (0)
......
include/util/tarray.h
浏览文件 @ 73a3f9bb
......@@ -205,6 +205,14 @@ SArray* taosArrayDup(const SArray* pSrc);
*/
void taosArrayClear(SArray* pArray);
/**
* clear the array (remove all element)
* @param pArray
* @param fp
*/
void taosArrayClearEx(SArray* pArray, void (*fp)(void*));
/**
* destroy array list
* @param pArray
......
include/util/tdef.h
浏览文件 @ 73a3f9bb
......@@ -128,6 +128,13 @@ extern const int32_t TYPE_BYTES[15];
#define TSDB_INS_TABLE_QUERIES "queries"
#define TSDB_INS_TABLE_VNODES "vnodes"
#define TSDB_PERFORMANCE_SCHEMA_DB "performance_schema"
#define TSDB_PERFS_TABLE_CONNECTIONS "connections"
#define TSDB_PERFS_TABLE_QUERIES "queries"
#define TSDB_PERFS_TABLE_TOPICS "topics"
#define TSDB_PERFS_TABLE_CONSUMERS "consumers"
#define TSDB_PERFS_TABLE_SUBSCRIBES "subscribes"
#define TSDB_INDEX_TYPE_SMA "SMA"
#define TSDB_INDEX_TYPE_FULLTEXT "FULLTEXT"
......
source/client/inc/clientInt.h
浏览文件 @ 73a3f9bb
......@@ -45,11 +45,6 @@ extern "C" {
#define HEARTBEAT_INTERVAL 1500 // ms
enum {
CONN_TYPE__QUERY = 1,
CONN_TYPE__TMQ,
};
typedef struct SAppInstInfo SAppInstInfo;
typedef struct {
......@@ -84,8 +79,8 @@ typedef struct {
TdThread thread;
TdThreadMutex lock; // used when app init and cleanup
SArray* appHbMgrs; // SArray<SAppHbMgr*> one for each cluster
FHbReqHandle reqHandle[HEARTBEAT_TYPE_MAX];
FHbRspHandle rspHandle[HEARTBEAT_TYPE_MAX];
FHbReqHandle reqHandle[CONN_TYPE__MAX];
FHbRspHandle rspHandle[CONN_TYPE__MAX];
} SClientHbMgr;
typedef struct SQueryExecMetric {
......@@ -144,6 +139,7 @@ typedef struct STscObj {
TdThreadMutex mutex; // used to protect the operation on db
int32_t numOfReqs; // number of sqlObj bound to this connection
SAppInstInfo* pAppInfo;
SHashObj* pRequests;
} STscObj;
typedef struct SResultColumn {
......@@ -256,11 +252,15 @@ int taos_init();
void* createTscObj(const char* user, const char* auth, const char* db, SAppInstInfo* pAppInfo);
void destroyTscObj(void* pObj);
STscObj *acquireTscObj(int64_t rid);
int32_t releaseTscObj(int64_t rid);
uint64_t generateRequestId();
void* createRequest(STscObj* pObj, __taos_async_fn_t fp, void* param, int32_t type);
void destroyRequest(SRequestObj* pRequest);
SRequestObj *acquireRequest(int64_t rid);
int32_t releaseRequest(int64_t rid);
char* getDbOfConnection(STscObj* pObj);
void setConnectionDB(STscObj* pTscObj, const char* db);
......@@ -302,7 +302,7 @@ SAppHbMgr* appHbMgrInit(SAppInstInfo* pAppInstInfo, char* key);
void appHbMgrCleanup(void);
// conn level
int hbRegisterConn(SAppHbMgr* pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType);
int hbRegisterConn(SAppHbMgr *pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType);
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);
......
source/client/inc/clientStmt.h 0 → 100644
浏览文件 @ 73a3f9bb
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_CLIENTSTMT_H
#define TDENGINE_CLIENTSTMT_H
#ifdef __cplusplus
extern "C" {
#endif
typedef enum {
STMT_TYPE_INSERT = 1,
STMT_TYPE_MULTI_INSERT,
STMT_TYPE_QUERY,
} STMT_TYPE;
typedef struct STscStmt {
STMT_TYPE type;
//int16_t last;
//STscObj* taos;
//SSqlObj* pSql;
//SMultiTbStmt mtb;
//SNormalStmt normal;
//int numOfRows;
} STscStmt;
#define STMT_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define STMT_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define STMT_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
TAOS_STMT *stmtInit(TAOS *taos);
int stmtClose(TAOS_STMT *stmt);
int stmtExec(TAOS_STMT *stmt);
char *stmtErrstr(TAOS_STMT *stmt);
int stmtAffectedRows(TAOS_STMT *stmt);
int stmtBind(TAOS_STMT *stmt, TAOS_BIND *bind);
int stmtPrepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
int stmtSetTbNameTags(TAOS_STMT *stmt, const char *name, TAOS_BIND *tags);
int stmtIsInsert(TAOS_STMT *stmt, int *insert);
int stmtGetParamNum(TAOS_STMT *stmt, int *nums);
int stmtAddBatch(TAOS_STMT *stmt);
TAOS_RES *stmtUseResult(TAOS_STMT *stmt);
int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_CLIENTSTMT_H
source/client/src/clientEnv.c
浏览文件 @ 73a3f9bb
......@@ -37,7 +37,8 @@ static TdThreadOnce tscinit = PTHREAD_ONCE_INIT;
volatile int32_t tscInitRes = 0;
static void registerRequest(SRequestObj *pRequest) {
STscObj *pTscObj = (STscObj *)taosAcquireRef(clientConnRefPool, pRequest->pTscObj->id);
STscObj *pTscObj = acquireTscObj(pRequest->pTscObj->id);
assert(pTscObj != NULL);
// connection has been released already, abort creating request.
......@@ -69,7 +70,7 @@ static void deregisterRequest(SRequestObj *pRequest) {
tscDebug("0x%" PRIx64 " free Request from connObj: 0x%" PRIx64 ", reqId:0x%" PRIx64 " elapsed:%" PRIu64
" ms, current:%d, app current:%d",
pRequest->self, pTscObj->id, pRequest->requestId, duration/1000, num, currentInst);
taosReleaseRef(clientConnRefPool, pTscObj->id);
releaseTscObj(pTscObj->id);
}
// todo close the transporter properly
......@@ -107,12 +108,24 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
return pDnodeConn;
}
void closeAllRequests(SHashObj *pRequests) {
void *pIter = taosHashIterate(pRequests, NULL);
while (pIter != NULL) {
int64_t *rid = pIter;
releaseRequest(*rid);
pIter = taosHashIterate(pRequests, pIter);
}
}
void destroyTscObj(void *pObj) {
STscObj *pTscObj = pObj;
SClientHbKey connKey = {.connId = pTscObj->connId, .hbType = pTscObj->connType};
SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType};
hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey);
atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
closeAllRequests(pTscObj->pRequests);
tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, pTscObj->id, pTscObj->pAppInfo->numOfConns);
taosThreadMutexDestroy(&pTscObj->mutex);
taosMemoryFreeClear(pTscObj);
......@@ -125,6 +138,13 @@ void *createTscObj(const char *user, const char *auth, const char *db, SAppInstI
return NULL;
}
pObj->pRequests = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
if (NULL == pObj->pRequests) {
taosMemoryFree(pObj);
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
return NULL;
}
pObj->pAppInfo = pAppInfo;
tstrncpy(pObj->user, user, sizeof(pObj->user));
memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);
......@@ -140,6 +160,14 @@ void *createTscObj(const char *user, const char *auth, const char *db, SAppInstI
return pObj;
}
STscObj *acquireTscObj(int64_t rid) {
return (STscObj *)taosAcquireRef(clientConnRefPool, rid);
}
int32_t releaseTscObj(int64_t rid) {
return taosReleaseRef(clientConnRefPool, rid);
}
void *createRequest(STscObj *pObj, __taos_async_fn_t fp, void *param, int32_t type) {
assert(pObj != NULL);
......@@ -161,6 +189,7 @@ void *createRequest(STscObj *pObj, __taos_async_fn_t fp, void *param, int32_t ty
tsem_init(&pRequest->body.rspSem, 0, 0);
registerRequest(pRequest);
return pRequest;
}
......@@ -186,6 +215,8 @@ static void doDestroyRequest(void *p) {
assert(RID_VALID(pRequest->self));
taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self));
taosMemoryFreeClear(pRequest->msgBuf);
taosMemoryFreeClear(pRequest->sqlstr);
taosMemoryFreeClear(pRequest->pInfo);
......@@ -214,9 +245,18 @@ void destroyRequest(SRequestObj *pRequest) {
return;
}
taosReleaseRef(clientReqRefPool, pRequest->self);
taosRemoveRef(clientReqRefPool, pRequest->self);
}
SRequestObj *acquireRequest(int64_t rid) {
return (SRequestObj *)taosAcquireRef(clientReqRefPool, rid);
}
int32_t releaseRequest(int64_t rid) {
return taosReleaseRef(clientReqRefPool, rid);
}
void taos_init_imp(void) {
// In the APIs of other program language, taos_cleanup is not available yet.
// So, to make sure taos_cleanup will be invoked to clean up the allocated resource to suppress the valgrind warning.
......@@ -457,11 +497,18 @@ uint64_t generateRequestId() {
}
}
int64_t ts = taosGetTimestampMs();
uint64_t pid = taosGetPId();
int32_t val = atomic_add_fetch_32(&requestSerialId, 1);
uint64_t id = 0;
while (true) {
int64_t ts = taosGetTimestampMs();
uint64_t pid = taosGetPId();
int32_t val = atomic_add_fetch_32(&requestSerialId, 1);
uint64_t id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
if (id) {
break;
}
}
return id;
}
......
source/client/src/clientHb.c
浏览文件 @ 73a3f9bb
......@@ -14,6 +14,7 @@
*/
#include "catalog.h"
#include "scheduler.h"
#include "clientInt.h"
#include "clientLog.h"
#include "trpc.h"
......@@ -109,10 +110,36 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
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);
tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, pRsp->connKey.connType);
return TSDB_CODE_SUCCESS;
}
if (pRsp->query) {
STscObj *pTscObj = (STscObj *)acquireTscObj(pRsp->connKey.tscRid);
if (NULL == pTscObj) {
tscDebug("tscObj rid %" PRIx64 " not exist", pRsp->connKey.tscRid);
} else {
updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &pRsp->query->epSet);
pTscObj->connId = pRsp->query->connId;
if (pRsp->query->killRid) {
SRequestObj *pRequest = acquireRequest(pRsp->query->killRid);
if (NULL == pRequest) {
tscDebug("request 0x%" PRIx64 " not exist to kill", pRsp->query->killRid);
} else {
taos_stop_query((TAOS_RES *)pRequest);
releaseRequest(pRsp->query->killRid);
}
}
if (pRsp->query->killConnection) {
taos_close(pTscObj);
}
releaseTscObj(pRsp->connKey.tscRid);
}
}
int32_t kvNum = pRsp->info ? taosArrayGetSize(pRsp->info) : 0;
tscDebug("hb got %d rsp kv", kvNum);
......@@ -197,7 +224,7 @@ static int32_t hbAsyncCallBack(void *param, const SDataBuf *pMsg, int32_t code)
for (int32_t i = 0; i < rspNum; ++i) {
SClientHbRsp *rsp = taosArrayGet(pRsp.rsps, i);
code = (*clientHbMgr.rspHandle[rsp->connKey.hbType])((*pInst)->pAppHbMgr, rsp);
code = (*clientHbMgr.rspHandle[rsp->connKey.connType])((*pInst)->pAppHbMgr, rsp);
if (code) {
break;
}
......@@ -208,6 +235,97 @@ static int32_t hbAsyncCallBack(void *param, const SDataBuf *pMsg, int32_t code)
return code;
}
int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
int64_t now = taosGetTimestampUs();
SQueryDesc desc = {0};
int32_t code = 0;
void *pIter = taosHashIterate(pObj->pRequests, NULL);
while (pIter != NULL) {
int64_t *rid = pIter;
SRequestObj *pRequest = acquireRequest(*rid);
if (NULL == pRequest) {
continue;
}
tstrncpy(desc.sql, pRequest->sqlstr, sizeof(desc.sql));
desc.stime = pRequest->metric.start;
desc.queryId = pRequest->requestId;
desc.useconds = now - pRequest->metric.start;
desc.reqRid = pRequest->self;
desc.pid = hbBasic->pid;
taosGetFqdn(desc.fqdn);
desc.subPlanNum = pRequest->body.pDag ? pRequest->body.pDag->numOfSubplans : 0;
if (desc.subPlanNum) {
desc.subDesc = taosArrayInit(desc.subPlanNum, sizeof(SQuerySubDesc));
if (NULL == desc.subDesc) {
releaseRequest(*rid);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
code = schedulerGetTasksStatus(pRequest->body.queryJob, desc.subDesc);
if (code) {
taosArrayDestroy(desc.subDesc);
desc.subDesc = NULL;
}
}
releaseRequest(*rid);
taosArrayPush(hbBasic->queryDesc, &desc);
pIter = taosHashIterate(pObj->pRequests, pIter);
}
return TSDB_CODE_SUCCESS;
}
int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
STscObj *pTscObj = (STscObj *)acquireTscObj(connKey->tscRid);
if (NULL == pTscObj) {
tscWarn("tscObj rid %" PRIx64 " not exist", connKey->tscRid);
return TSDB_CODE_QRY_APP_ERROR;
}
int32_t numOfQueries = pTscObj->pRequests ? taosHashGetSize(pTscObj->pRequests) : 0;
if (numOfQueries <= 0) {
releaseTscObj(connKey->tscRid);
tscDebug("no queries on connection");
return TSDB_CODE_QRY_APP_ERROR;
}
SQueryHbReqBasic *hbBasic = (SQueryHbReqBasic *)taosMemoryCalloc(1, sizeof(SQueryHbReqBasic));
if (NULL == hbBasic) {
tscError("calloc %d failed", (int32_t)sizeof(SQueryHbReqBasic));
releaseTscObj(connKey->tscRid);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
hbBasic->queryDesc = taosArrayInit(numOfQueries, sizeof(SQueryDesc));
if (NULL == hbBasic->queryDesc) {
tscWarn("taosArrayInit %d queryDesc failed", numOfQueries);
releaseTscObj(connKey->tscRid);
taosMemoryFree(hbBasic);
return TSDB_CODE_QRY_OUT_OF_MEMORY;
}
hbBasic->connId = pTscObj->connId;
hbBasic->pid = taosGetPId();
taosGetAppName(hbBasic->app, NULL);
int32_t code = hbBuildQueryDesc(hbBasic, pTscObj);
if (code) {
releaseTscObj(connKey->tscRid);
taosMemoryFree(hbBasic);
return code;
}
req->query = hbBasic;
releaseTscObj(connKey->tscRid);
return TSDB_CODE_SUCCESS;
}
int32_t hbGetExpiredDBInfo(SClientHbKey *connKey, struct SCatalog *pCatalog, SClientHbReq *req) {
SDbVgVersion *dbs = NULL;
uint32_t dbNum = 0;
......@@ -286,6 +404,8 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
return code;
}
hbGetQueryBasicInfo(connKey, req);
code = hbGetExpiredDBInfo(connKey, pCatalog, req);
if (TSDB_CODE_SUCCESS != code) {
return code;
......@@ -300,11 +420,11 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
}
void hbMgrInitMqHbHandle() {
clientHbMgr.reqHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbReqHandle;
clientHbMgr.reqHandle[HEARTBEAT_TYPE_MQ] = hbMqHbReqHandle;
clientHbMgr.reqHandle[CONN_TYPE__QUERY] = hbQueryHbReqHandle;
clientHbMgr.reqHandle[CONN_TYPE__TMQ] = hbMqHbReqHandle;
clientHbMgr.rspHandle[HEARTBEAT_TYPE_QUERY] = hbQueryHbRspHandle;
clientHbMgr.rspHandle[HEARTBEAT_TYPE_MQ] = hbMqHbRspHandle;
clientHbMgr.rspHandle[CONN_TYPE__QUERY] = hbQueryHbRspHandle;
clientHbMgr.rspHandle[CONN_TYPE__TMQ] = hbMqHbRspHandle;
}
static FORCE_INLINE void hbMgrInitHandle() {
......@@ -317,6 +437,11 @@ void hbFreeReq(void *req) {
tFreeReqKvHash(pReq->info);
}
void hbClearClientHbReq(SClientHbReq *pReq) {
pReq->query = NULL;
pReq->info = NULL;
}
SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
SClientHbBatchReq *pBatchReq = taosMemoryCalloc(1, sizeof(SClientHbBatchReq));
if (pBatchReq == NULL) {
......@@ -333,22 +458,23 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pOneReq->connKey, sizeof(SClientHbKey));
if (info) {
code = (*clientHbMgr.reqHandle[pOneReq->connKey.hbType])(&pOneReq->connKey, info->param, pOneReq);
code = (*clientHbMgr.reqHandle[pOneReq->connKey.connType])(&pOneReq->connKey, info->param, pOneReq);
if (code) {
taosHashCancelIterate(pAppHbMgr->activeInfo, pIter);
break;
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
continue;
}
}
taosArrayPush(pBatchReq->reqs, pOneReq);
hbClearClientHbReq(pOneReq);
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
}
if (code) {
taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
taosMemoryFreeClear(pBatchReq);
}
// if (code) {
// taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
// taosMemoryFreeClear(pBatchReq);
// }
return pBatchReq;
}
......@@ -523,13 +649,13 @@ int hbMgrInit() {
hbMgrInitHandle();
// init backgroud thread
hbCreateThread();
//hbCreateThread();
return 0;
}
void hbMgrCleanUp() {
hbStopThread();
//hbStopThread();
// destroy all appHbMgr
int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0);
......@@ -549,7 +675,7 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *
if (data != NULL) {
return 0;
}
SClientHbReq hbReq;
SClientHbReq hbReq = {0};
hbReq.connKey = connKey;
hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
......@@ -566,22 +692,22 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *
return 0;
}
int hbRegisterConn(SAppHbMgr *pAppHbMgr, int32_t connId, int64_t clusterId, int32_t hbType) {
int hbRegisterConn(SAppHbMgr *pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType) {
SClientHbKey connKey = {
.connId = connId,
.hbType = hbType,
.tscRid = tscRefId,
.connType = connType,
};
SHbConnInfo info = {0};
switch (hbType) {
case HEARTBEAT_TYPE_QUERY: {
switch (connType) {
case CONN_TYPE__QUERY: {
int64_t *pClusterId = taosMemoryMalloc(sizeof(int64_t));
*pClusterId = clusterId;
info.param = pClusterId;
return hbRegisterConnImpl(pAppHbMgr, connKey, &info);
}
case HEARTBEAT_TYPE_MQ: {
case CONN_TYPE__TMQ: {
return 0;
}
default:
......
source/client/src/clientImpl.c
浏览文件 @ 73a3f9bb
......@@ -132,6 +132,13 @@ int32_t buildRequest(STscObj* pTscObj, const char* sql, int sqlLen, SRequestObj*
(*pRequest)->sqlstr[sqlLen] = 0;
(*pRequest)->sqlLen = sqlLen;
if (taosHashPut(pTscObj->pRequests, &(*pRequest)->self, sizeof((*pRequest)->self), &(*pRequest)->self, sizeof((*pRequest)->self))) {
destroyRequest(*pRequest);
*pRequest = NULL;
tscError("put request to request hash failed");
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
tscDebugL("0x%" PRIx64 " SQL: %s, reqId:0x%" PRIx64, (*pRequest)->self, (*pRequest)->sqlstr, (*pRequest)->requestId);
return TSDB_CODE_SUCCESS;
}
......@@ -447,7 +454,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t
taos_close(pTscObj);
pTscObj = NULL;
} else {
tscDebug("0x%" PRIx64 " connection is opening, connId:%d, dnodeConn:%p, reqId:0x%" PRIx64, pTscObj->id,
tscDebug("0x%" PRIx64 " connection is opening, connId:%u, dnodeConn:%p, reqId:0x%" PRIx64, pTscObj->id,
pTscObj->connId, pTscObj->pAppInfo->pTransporter, pRequest->requestId);
destroyRequest(pRequest);
}
......
source/client/src/clientMain.c
浏览文件 @ 73a3f9bb
......@@ -14,7 +14,9 @@
*/
#include "catalog.h"
#include "scheduler.h"
#include "clientInt.h"
#include "clientStmt.h"
#include "clientLog.h"
#include "os.h"
#include "query.h"
......@@ -66,6 +68,7 @@ void taos_cleanup(void) {
rpcCleanup();
catalogDestroy();
schedulerDestroy();
taosCloseLog();
tscInfo("all local resources released");
......@@ -98,7 +101,7 @@ void taos_close(TAOS *taos) {
STscObj *pTscObj = (STscObj *)taos;
tscDebug("0x%" PRIx64 " try to close connection, numOfReq:%d", pTscObj->id, pTscObj->numOfReqs);
/*taosRemoveRef(clientConnRefPool, pTscObj->id);*/
taosRemoveRef(clientConnRefPool, pTscObj->id);
}
int taos_errno(TAOS_RES *tres) {
......@@ -400,7 +403,7 @@ void taos_stop_query(TAOS_RES *res) {
return;
}
// scheduleCancelJob(pRequest->body.pQueryJob);
schedulerFreeJob(pRequest->body.queryJob);
}
bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
......@@ -565,76 +568,149 @@ int taos_load_table_info(TAOS *taos, const char *tableNameList) {
}
TAOS_STMT *taos_stmt_init(TAOS *taos) {
// TODO
return NULL;
if (taos == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
return stmtInit(taos);
}
int taos_stmt_close(TAOS_STMT *stmt) {
// TODO
return -1;
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtClose(stmt);
}
int taos_stmt_execute(TAOS_STMT *stmt) {
// TODO
return -1;
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtExec(stmt);
}
char *taos_stmt_errstr(TAOS_STMT *stmt) {
// TODO
return NULL;
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
return stmtErrstr(stmt);
}
int taos_stmt_affected_rows(TAOS_STMT *stmt) {
// TODO
return -1;
}
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return 0;
}
TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
// TODO
return NULL;
return stmtAffectedRows(stmt);
}
int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_BIND *bind) {
// TODO
return -1;
if (stmt == NULL || bind == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtBind(stmt, bind);
}
int taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
// TODO
return -1;
if (stmt == NULL || sql == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtPrepare(stmt, sql, length);
}
int taos_stmt_set_tbname_tags(TAOS_STMT *stmt, const char *name, TAOS_BIND *tags) {
// TODO
return -1;
if (stmt == NULL || name == NULL || tags == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtSetTbNameTags(stmt, name, tags);
}
int taos_stmt_set_tbname(TAOS_STMT *stmt, const char *name) {
// TODO
return -1;
if (stmt == NULL || name == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtSetTbNameTags(stmt, name, NULL);
}
int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert) {
// TODO
return -1;
if (stmt == NULL || insert == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtIsInsert(stmt, insert);
}
int taos_stmt_num_params(TAOS_STMT *stmt, int *nums) {
// TODO
return -1;
if (stmt == NULL || nums == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtGetParamNum(stmt, nums);
}
int taos_stmt_add_batch(TAOS_STMT *stmt) {
// TODO
return -1;
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtAddBatch(stmt);
}
TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
// TODO
return NULL;
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
return stmtUseResult(stmt);
}
int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
if (stmt == NULL || bind == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
return stmtBindBatch(stmt, bind);
}
TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
// TODO
return -1;
return NULL;
}
source/client/src/clientMsgHandler.c
浏览文件 @ 73a3f9bb
......@@ -71,7 +71,7 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pTscObj->connType = connectRsp.connType;
hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, connectRsp.connId, connectRsp.clusterId, connectRsp.connType);
hbRegisterConn(pTscObj->pAppInfo->pAppHbMgr, pTscObj->id, connectRsp.clusterId, connectRsp.connType);
// pRequest->body.resInfo.pRspMsg = pMsg->pData;
tscDebug("0x%" PRIx64 " clusterId:%" PRId64 ", totalConn:%" PRId64, pRequest->requestId, connectRsp.clusterId,
......
source/client/src/clientStmt.c 0 → 100644
浏览文件 @ 73a3f9bb
#include "clientInt.h"
#include "clientLog.h"
#include "clientStmt.h"
#include "tdef.h"
TAOS_STMT *stmtInit(TAOS *taos) {
STscObj* pObj = (STscObj*)taos;
STscStmt* pStmt = NULL;
#if 0
pStmt = taosMemoryCalloc(1, sizeof(STscStmt));
if (pStmt == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscError("failed to allocate memory for statement");
return NULL;
}
pStmt->taos = pObj;
SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
if (pSql == NULL) {
free(pStmt);
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscError("failed to allocate memory for statement");
return NULL;
}
if (TSDB_CODE_SUCCESS != tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
free(pSql);
free(pStmt);
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
tscError("failed to malloc payload buffer");
return NULL;
}
tsem_init(&pSql->rspSem, 0, 0);
pSql->signature = pSql;
pSql->pTscObj = pObj;
pSql->maxRetry = TSDB_MAX_REPLICA;
pStmt->pSql = pSql;
pStmt->last = STMT_INIT;
pStmt->numOfRows = 0;
registerSqlObj(pSql);
#endif
return pStmt;
}
int stmtClose(TAOS_STMT *stmt) {
return TSDB_CODE_SUCCESS;
}
int stmtExec(TAOS_STMT *stmt) {
return TSDB_CODE_SUCCESS;
}
char *stmtErrstr(TAOS_STMT *stmt) {
return NULL;
}
int stmtAffectedRows(TAOS_STMT *stmt) {
return TSDB_CODE_SUCCESS;
}
int stmtBind(TAOS_STMT *stmt, TAOS_BIND *bind) {
return TSDB_CODE_SUCCESS;
}
int stmtPrepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
return TSDB_CODE_SUCCESS;
}
int stmtSetTbNameTags(TAOS_STMT *stmt, const char *name, TAOS_BIND *tags) {
return TSDB_CODE_SUCCESS;
}
int stmtIsInsert(TAOS_STMT *stmt, int *insert) {
return TSDB_CODE_SUCCESS;
}
int stmtGetParamNum(TAOS_STMT *stmt, int *nums) {
return TSDB_CODE_SUCCESS;
}
int stmtAddBatch(TAOS_STMT *stmt) {
return TSDB_CODE_SUCCESS;
}
TAOS_RES *stmtUseResult(TAOS_STMT *stmt) {
return NULL;
}
int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
return TSDB_CODE_SUCCESS;
}
source/common/src/tmsg.c
浏览文件 @ 73a3f9bb
......@@ -134,6 +134,42 @@ void *taosDecodeSEpSet(void *buf, SEpSet *pEp) {
static int32_t tSerializeSClientHbReq(SCoder *pEncoder, const SClientHbReq *pReq) {
if (tEncodeSClientHbKey(pEncoder, &pReq->connKey) < 0) return -1;
if (pReq->connKey.connType == CONN_TYPE__QUERY) {
int32_t queryNum = 0;
if (pReq->query) {
queryNum = 1;
if (tEncodeI32(pEncoder, queryNum) < 0) return -1;
if (tEncodeU32(pEncoder, pReq->query->connId) < 0) return -1;
if (tEncodeI32(pEncoder, pReq->query->pid) < 0) return -1;
if (tEncodeCStr(pEncoder, pReq->query->app) < 0) return -1;
int32_t num = taosArrayGetSize(pReq->query->queryDesc);
if (tEncodeI32(pEncoder, num) < 0) return -1;
for (int32_t i = 0; i < num; ++i) {
SQueryDesc *desc = taosArrayGet(pReq->query->queryDesc, i);
if (tEncodeCStr(pEncoder, desc->sql) < 0) return -1;
if (tEncodeU64(pEncoder, desc->queryId) < 0) return -1;
if (tEncodeI64(pEncoder, desc->useconds) < 0) return -1;
if (tEncodeI64(pEncoder, desc->stime) < 0) return -1;
if (tEncodeI64(pEncoder, desc->reqRid) < 0) return -1;
if (tEncodeI32(pEncoder, desc->pid) < 0) return -1;
if (tEncodeCStr(pEncoder, desc->fqdn) < 0) return -1;
if (tEncodeI32(pEncoder, desc->subPlanNum) < 0) return -1;
int32_t snum = desc->subDesc ? taosArrayGetSize(desc->subDesc) : 0;
if (tEncodeI32(pEncoder, snum) < 0) return -1;
for (int32_t m = 0; m < snum; ++m) {
SQuerySubDesc *sDesc = taosArrayGet(desc->subDesc, m);
if (tEncodeI64(pEncoder, sDesc->tid) < 0) return -1;
if (tEncodeI32(pEncoder, sDesc->status) < 0) return -1;
}
}
} else {
if (tEncodeI32(pEncoder, queryNum) < 0) return -1;
}
}
int32_t kvNum = taosHashGetSize(pReq->info);
if (tEncodeI32(pEncoder, kvNum) < 0) return -1;
void *pIter = taosHashIterate(pReq->info, NULL);
......@@ -149,6 +185,53 @@ static int32_t tSerializeSClientHbReq(SCoder *pEncoder, const SClientHbReq *pReq
static int32_t tDeserializeSClientHbReq(SCoder *pDecoder, SClientHbReq *pReq) {
if (tDecodeSClientHbKey(pDecoder, &pReq->connKey) < 0) return -1;
if (pReq->connKey.connType == CONN_TYPE__QUERY) {
int32_t queryNum = 0;
if (tDecodeI32(pDecoder, &queryNum) < 0) return -1;
if (queryNum) {
pReq->query = taosMemoryCalloc(1, sizeof(*pReq->query));
if (NULL == pReq->query) return -1;
if (tDecodeU32(pDecoder, &pReq->query->connId) < 0) return -1;
if (tDecodeI32(pDecoder, &pReq->query->pid) < 0) return -1;
if (tDecodeCStrTo(pDecoder, pReq->query->app) < 0) return -1;
int32_t num = 0;
if (tDecodeI32(pDecoder, &num) < 0) return -1;
if (num > 0) {
pReq->query->queryDesc = taosArrayInit(num, sizeof(SQueryDesc));
if (NULL == pReq->query->queryDesc) return -1;
for (int32_t i = 0; i < num; ++i) {
SQueryDesc desc = {0};
if (tDecodeCStrTo(pDecoder, desc.sql) < 0) return -1;
if (tDecodeU64(pDecoder, &desc.queryId) < 0) return -1;
if (tDecodeI64(pDecoder, &desc.useconds) < 0) return -1;
if (tDecodeI64(pDecoder, &desc.stime) < 0) return -1;
if (tDecodeI64(pDecoder, &desc.reqRid) < 0) return -1;
if (tDecodeI32(pDecoder, &desc.pid) < 0) return -1;
if (tDecodeCStrTo(pDecoder, desc.fqdn) < 0) return -1;
if (tDecodeI32(pDecoder, &desc.subPlanNum) < 0) return -1;
int32_t snum = 0;
if (tDecodeI32(pDecoder, &snum) < 0) return -1;
if (snum > 0) {
desc.subDesc = taosArrayInit(snum, sizeof(SQuerySubDesc));
if (NULL == desc.subDesc) return -1;
for (int32_t m = 0; m < snum; ++m) {
SQuerySubDesc sDesc = {0};
if (tDecodeI64(pDecoder, &sDesc.tid) < 0) return -1;
if (tDecodeI32(pDecoder, &sDesc.status) < 0) return -1;
taosArrayPush(desc.subDesc, &sDesc);
}
}
taosArrayPush(pReq->query->queryDesc, &desc);
}
}
}
}
int32_t kvNum = 0;
if (tDecodeI32(pDecoder, &kvNum) < 0) return -1;
if (pReq->info == NULL) {
......@@ -168,6 +251,20 @@ static int32_t tSerializeSClientHbRsp(SCoder *pEncoder, const SClientHbRsp *pRsp
if (tEncodeSClientHbKey(pEncoder, &pRsp->connKey) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->status) < 0) return -1;
int32_t queryNum = 0;
if (pRsp->query) {
queryNum = 1;
if (tEncodeI32(pEncoder, queryNum) < 0) return -1;
if (tEncodeU32(pEncoder, pRsp->query->connId) < 0) return -1;
if (tEncodeU64(pEncoder, pRsp->query->killRid) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->query->totalDnodes) < 0) return -1;
if (tEncodeI32(pEncoder, pRsp->query->onlineDnodes) < 0) return -1;
if (tEncodeI8(pEncoder, pRsp->query->killConnection) < 0) return -1;
if (tEncodeSEpSet(pEncoder, &pRsp->query->epSet) < 0) return -1;
} else {
if (tEncodeI32(pEncoder, queryNum) < 0) return -1;
}
int32_t kvNum = taosArrayGetSize(pRsp->info);
if (tEncodeI32(pEncoder, kvNum) < 0) return -1;
for (int32_t i = 0; i < kvNum; i++) {
......@@ -182,6 +279,19 @@ static int32_t tDeserializeSClientHbRsp(SCoder *pDecoder, SClientHbRsp *pRsp) {
if (tDecodeSClientHbKey(pDecoder, &pRsp->connKey) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->status) < 0) return -1;
int32_t queryNum = 0;
if (tDecodeI32(pDecoder, &queryNum) < 0) return -1;
if (queryNum) {
pRsp->query = taosMemoryCalloc(1, sizeof(*pRsp->query));
if (NULL == pRsp->query) return -1;
if (tDecodeU32(pDecoder, &pRsp->query->connId) < 0) return -1;
if (tDecodeU64(pDecoder, &pRsp->query->killRid) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->query->totalDnodes) < 0) return -1;
if (tDecodeI32(pDecoder, &pRsp->query->onlineDnodes) < 0) return -1;
if (tDecodeI8(pDecoder, &pRsp->query->killConnection) < 0) return -1;
if (tDecodeSEpSet(pDecoder, &pRsp->query->epSet) < 0) return -1;
}
int32_t kvNum = 0;
if (tDecodeI32(pDecoder, &kvNum) < 0) return -1;
pRsp->info = taosArrayInit(kvNum, sizeof(SKv));
......@@ -224,8 +334,9 @@ int32_t tDeserializeSClientHbBatchReq(void *buf, int32_t bufLen, SClientHbBatchR
int32_t reqNum = 0;
if (tDecodeI32(&decoder, &reqNum) < 0) return -1;
if (pBatchReq->reqs == NULL) {
if (reqNum > 0) {
pBatchReq->reqs = taosArrayInit(reqNum, sizeof(SClientHbReq));
if (NULL == pBatchReq->reqs) return -1;
}
for (int32_t i = 0; i < reqNum; i++) {
SClientHbReq req = {0};
......@@ -2567,7 +2678,7 @@ int32_t tSerializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->acctId) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->clusterId) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->connId) < 0) return -1;
if (tEncodeU32(&encoder, pRsp->connId) < 0) return -1;
if (tEncodeI8(&encoder, pRsp->superUser) < 0) return -1;
if (tEncodeI8(&encoder, pRsp->connType) < 0) return -1;
if (tEncodeSEpSet(&encoder, &pRsp->epSet) < 0) return -1;
......@@ -2586,7 +2697,7 @@ int32_t tDeserializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->acctId) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->clusterId) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->connId) < 0) return -1;
if (tDecodeU32(&decoder, &pRsp->connId) < 0) return -1;
if (tDecodeI8(&decoder, &pRsp->superUser) < 0) return -1;
if (tDecodeI8(&decoder, &pRsp->connType) < 0) return -1;
if (tDecodeSEpSet(&decoder, &pRsp->epSet) < 0) return -1;
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ 73a3f9bb
......@@ -38,6 +38,10 @@ extern "C" {
#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 SYSTABLE_SCH_TABLE_NAME_LEN ((TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
#define SYSTABLE_SCH_DB_NAME_LEN ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
#define SYSTABLE_SCH_COL_NAME_LEN ((TSDB_COL_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
typedef int32_t (*MndMsgFp)(SNodeMsg *pMsg);
typedef int32_t (*MndInitFp)(SMnode *pMnode);
typedef void (*MndCleanupFp)(SMnode *pMnode);
......@@ -74,7 +78,6 @@ typedef struct {
} SShowMgmt;
typedef struct {
int32_t connId;
SCacheObj *cache;
} SProfileMgmt;
......@@ -118,6 +121,7 @@ struct SMnode {
STelemMgmt telemMgmt;
SSyncMgmt syncMgmt;
SHashObj *infosMeta;
SHashObj *perfsMeta;
SGrantInfo grant;
MndMsgFp msgFp[TDMT_MAX];
SMsgCb msgCb;
......
source/dnode/mnode/impl/inc/mndPerfSchema.h 0 → 100644
浏览文件 @ 73a3f9bb
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_MND_PERF_SCHEMA_H_
#define _TD_MND_PERF_SCHEMA_H_
#include "mndInt.h"
#ifdef __cplusplus
extern "C" {
#endif
typedef struct SPerfsTableSchema {
char *name;
int32_t type;
int32_t bytes;
} SPerfsTableSchema;
typedef struct SPerfsTableMeta {
char *name;
const SPerfsTableSchema *schema;
int32_t colNum;
} SPerfsTableMeta;
int32_t mndBuildPerfsTableSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp);
int32_t mndInitPerfs(SMnode *pMnode);
void mndCleanupPerfs(SMnode *pMnode);
#ifdef __cplusplus
}
#endif
#endif /*_TD_MND_PERF_SCHEMA_H_*/
source/dnode/mnode/impl/src/mndDb.c
浏览文件 @ 73a3f9bb
......@@ -1128,6 +1128,8 @@ static int32_t mndProcessUseDbReq(SNodeMsg *pReq) {
if (taosArrayGetSize(usedbRsp.pVgroupInfos) <= 0) {
terrno = TSDB_CODE_MND_DB_NOT_EXIST;
} else {
code = 0;
}
} else {
usedbRsp.vgVersion = usedbReq.vgVersion;
......
source/dnode/mnode/impl/src/mndInfoSchema.c
浏览文件 @ 73a3f9bb
......@@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE
#include "mndInfoSchema.h"
#include "mndInt.h"
#define SYSTABLE_SCH_TABLE_NAME_LEN ((TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
#define SYSTABLE_SCH_DB_NAME_LEN ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
......
source/dnode/mnode/impl/src/mndPerfSchema.c 0 → 100644
浏览文件 @ 73a3f9bb
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#define _DEFAULT_SOURCE
#include "mndPerfSchema.h"
#include "mndInt.h"
//!!!! Note: only APPEND columns in below tables, NO insert !!!!
static const SPerfsTableSchema connectionsSchema[] = {
{.name = "conn_id", .bytes = 4, .type = TSDB_DATA_TYPE_UINT},
{.name = "user", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "program", .bytes = TSDB_APP_NAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "end_point", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "login_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "last_access", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
};
static const SPerfsTableSchema queriesSchema[] = {
{.name = "query_id", .bytes = 4, .type = TSDB_DATA_TYPE_UBIGINT},
{.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "user", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "fqdn", .bytes = TSDB_FQDN_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "exec_time", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
{.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "sub_queries", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "sub_query_info", .bytes = TSDB_SHOW_SUBQUERY_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
};
static const SPerfsTableSchema topicSchema[] = {
{.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
{.name = "row_len", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
};
static const SPerfsTableSchema consumerSchema[] = {
{.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
// ep
// up time
// topics
};
static const SPerfsTableSchema subscribeSchema[] = {
{.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
{.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
};
static const SPerfsTableMeta perfsMeta[] = {
{TSDB_PERFS_TABLE_CONNECTIONS, connectionsSchema, tListLen(connectionsSchema)},
{TSDB_PERFS_TABLE_QUERIES, queriesSchema, tListLen(queriesSchema)},
{TSDB_PERFS_TABLE_TOPICS, topicSchema, tListLen(topicSchema)},
{TSDB_PERFS_TABLE_CONSUMERS, consumerSchema, tListLen(consumerSchema)},
{TSDB_PERFS_TABLE_SUBSCRIBES, subscribeSchema, tListLen(subscribeSchema)},
};
// connection/application/
int32_t mndInitPerfsTableSchema(const SPerfsTableSchema *pSrc, int32_t colNum, SSchema **pDst) {
SSchema *schema = taosMemoryCalloc(colNum, sizeof(SSchema));
if (NULL == schema) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
for (int32_t i = 0; i < colNum; ++i) {
strcpy(schema[i].name, pSrc[i].name);
schema[i].type = pSrc[i].type;
schema[i].colId = i + 1;
schema[i].bytes = pSrc[i].bytes;
}
*pDst = schema;
return TSDB_CODE_SUCCESS;
}
int32_t mndPerfsInitMeta(SHashObj *hash) {
STableMetaRsp meta = {0};
strcpy(meta.dbFName, TSDB_INFORMATION_SCHEMA_DB);
meta.tableType = TSDB_SYSTEM_TABLE;
meta.sversion = 1;
meta.tversion = 1;
for (int32_t i = 0; i < tListLen(perfsMeta); ++i) {
strcpy(meta.tbName, perfsMeta[i].name);
meta.numOfColumns = perfsMeta[i].colNum;
if (mndInitPerfsTableSchema(perfsMeta[i].schema, perfsMeta[i].colNum, &meta.pSchemas)) {
return -1;
}
if (taosHashPut(hash, meta.tbName, strlen(meta.tbName), &meta, sizeof(meta))) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t mndBuildPerfsTableSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {
if (NULL == pMnode->perfsMeta) {
terrno = TSDB_CODE_MND_NOT_READY;
return -1;
}
STableMetaRsp *meta = (STableMetaRsp *)taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));
if (NULL == meta) {
mError("invalid performance schema table name:%s", tbName);
terrno = TSDB_CODE_MND_INVALID_INFOS_TBL;
return -1;
}
*pRsp = *meta;
pRsp->pSchemas = taosMemoryCalloc(meta->numOfColumns, sizeof(SSchema));
if (pRsp->pSchemas == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
pRsp->pSchemas = NULL;
return -1;
}
memcpy(pRsp->pSchemas, meta->pSchemas, meta->numOfColumns * sizeof(SSchema));
return 0;
}
int32_t mndInitPerfs(SMnode *pMnode) {
pMnode->perfsMeta = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (pMnode->perfsMeta == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
return mndPerfsInitMeta(pMnode->perfsMeta);
}
void mndCleanupPerfs(SMnode *pMnode) {
if (NULL == pMnode->perfsMeta) {
return;
}
void *pIter = taosHashIterate(pMnode->perfsMeta, NULL);
while (pIter) {
STableMetaRsp *meta = (STableMetaRsp *)pIter;
taosMemoryFreeClear(meta->pSchemas);
pIter = taosHashIterate(pMnode->perfsMeta, pIter);
}
taosHashCleanup(pMnode->perfsMeta);
pMnode->perfsMeta = NULL;
}
source/dnode/mnode/impl/src/mndProfile.c
浏览文件 @ 73a3f9bb
......@@ -24,7 +24,7 @@
#include "version.h"
typedef struct {
int32_t id;
uint32_t id;
int8_t connType;
char user[TSDB_USER_LEN];
char app[TSDB_APP_NAME_LEN]; // app name that invokes taosc
......@@ -35,15 +35,15 @@ typedef struct {
int8_t killed;
int64_t loginTimeMs;
int64_t lastAccessTimeMs;
int32_t queryId;
uint64_t killId;
int32_t numOfQueries;
SQueryDesc *pQueries;
SArray *pQueries; //SArray<SQueryDesc>
} SConnObj;
static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType, uint32_t ip, uint16_t port,
int32_t pid, const char *app, int64_t startTime);
static void mndFreeConn(SConnObj *pConn);
static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId);
static SConnObj *mndAcquireConn(SMnode *pMnode, uint32_t connId);
static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn);
static void *mndGetNextConn(SMnode *pMnode, SCacheIter *pIter);
static void mndCancelGetNextConn(SMnode *pMnode, void *pIter);
......@@ -91,8 +91,9 @@ static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType
int32_t pid, const char *app, int64_t startTime) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
int32_t connId = atomic_add_fetch_32(&pMgmt->connId, 1);
if (connId == 0) atomic_add_fetch_32(&pMgmt->connId, 1);
char connStr[255] = {0};
int32_t len = snprintf(connStr, sizeof(connStr), "%s%d%d%d%s", user, ip, port, pid, app);
int32_t connId = mndGenerateUid(connStr, len);
if (startTime == 0) startTime = taosGetTimestampMs();
SConnObj connObj = {.id = connId,
......@@ -104,7 +105,7 @@ static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType
.killed = 0,
.loginTimeMs = taosGetTimestampMs(),
.lastAccessTimeMs = 0,
.queryId = 0,
.killId = 0,
.numOfQueries = 0,
.pQueries = NULL};
......@@ -119,35 +120,35 @@ static SConnObj *mndCreateConn(SMnode *pMnode, const char *user, int8_t connType
mError("conn:%d, failed to put into cache since %s, user:%s", connId, user, terrstr());
return NULL;
} else {
mTrace("conn:%d, is created, data:%p user:%s", pConn->id, pConn, user);
mTrace("conn:%u, is created, data:%p user:%s", pConn->id, pConn, user);
return pConn;
}
}
static void mndFreeConn(SConnObj *pConn) {
taosMemoryFreeClear(pConn->pQueries);
mTrace("conn:%d, is destroyed, data:%p", pConn->id, pConn);
mTrace("conn:%u, is destroyed, data:%p", pConn->id, pConn);
}
static SConnObj *mndAcquireConn(SMnode *pMnode, int32_t connId) {
static SConnObj *mndAcquireConn(SMnode *pMnode, uint32_t connId) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &connId, sizeof(int32_t));
SConnObj *pConn = taosCacheAcquireByKey(pMgmt->cache, &connId, sizeof(connId));
if (pConn == NULL) {
mDebug("conn:%d, already destroyed", connId);
mDebug("conn:%u, already destroyed", connId);
return NULL;
}
int32_t keepTime = tsShellActivityTimer * 3;
pConn->lastAccessTimeMs = keepTime * 1000 + (uint64_t)taosGetTimestampMs();
mTrace("conn:%d, acquired from cache, data:%p", pConn->id, pConn);
mTrace("conn:%u, acquired from cache, data:%p", pConn->id, pConn);
return pConn;
}
static void mndReleaseConn(SMnode *pMnode, SConnObj *pConn) {
if (pConn == NULL) return;
mTrace("conn:%d, released from cache, data:%p", pConn->id, pConn);
mTrace("conn:%u, released from cache, data:%p", pConn->id, pConn);
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
......@@ -212,6 +213,8 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
goto CONN_OVER;
}
mndAcquireConn(pMnode, pConn->id);
SConnectRsp connectRsp = {0};
connectRsp.acctId = pUser->acctId;
connectRsp.superUser = pUser->superUser;
......@@ -232,7 +235,7 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
pReq->rspLen = contLen;
pReq->pRsp = pRsp;
mDebug("user:%s, login from %s, conn:%d, app:%s", pReq->user, ip, pConn->id, connReq.app);
mDebug("user:%s, login from %s:%d, conn:%u, app:%s", pReq->user, ip, pConn->port, pConn->id, connReq.app);
code = 0;
......@@ -245,22 +248,13 @@ CONN_OVER:
return code;
}
static int32_t mndSaveQueryStreamList(SConnObj *pConn, SHeartBeatReq *pReq) {
pConn->numOfQueries = 0;
int32_t numOfQueries = htonl(pReq->numOfQueries);
if (numOfQueries > 0) {
if (pConn->pQueries == NULL) {
pConn->pQueries = taosMemoryCalloc(sizeof(SQueryDesc), QUERY_SAVE_SIZE);
}
pConn->numOfQueries = TMIN(QUERY_SAVE_SIZE, numOfQueries);
static int32_t mndSaveQueryList(SConnObj *pConn, SQueryHbReqBasic *pBasic) {
taosArrayDestroyEx(pConn->pQueries, tFreeClientHbQueryDesc);
int32_t saveSize = pConn->numOfQueries * sizeof(SQueryDesc);
if (saveSize > 0 && pConn->pQueries != NULL) {
memcpy(pConn->pQueries, pReq->pData, saveSize);
}
}
pConn->pQueries = pBasic->queryDesc;
pBasic->queryDesc = NULL;
pConn->numOfQueries = pBasic->queryDesc ? taosArrayGetSize(pBasic->queryDesc) : 0;
return TSDB_CODE_SUCCESS;
}
......@@ -330,6 +324,111 @@ static SClientHbRsp *mndMqHbBuildRsp(SMnode *pMnode, SClientHbReq *pReq) {
return NULL;
}
static int32_t mndProcessQueryHeartBeat(SMnode *pMnode, SRpcMsg *pMsg, SClientHbReq *pHbReq, SClientHbBatchRsp *pBatchRsp) {
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SClientHbRsp hbRsp = {.connKey = pHbReq->connKey, .status = 0, .info = NULL, .query = NULL};
if (pHbReq->query) {
SQueryHbReqBasic *pBasic = pHbReq->query;
SRpcConnInfo connInfo = {0};
rpcGetConnInfo(pMsg->handle, &connInfo);
SConnObj *pConn = mndAcquireConn(pMnode, pBasic->connId);
if (pConn == NULL) {
pConn = mndCreateConn(pMnode, connInfo.user, CONN_TYPE__QUERY, connInfo.clientIp, connInfo.clientPort, pBasic->pid, pBasic->app, 0);
if (pConn == NULL) {
mError("user:%s, conn:%u is freed and failed to create new since %s", connInfo.user, pBasic->connId, terrstr());
return -1;
} else {
mDebug("user:%s, conn:%u is freed and create a new conn:%u", connInfo.user, pBasic->connId, pConn->id);
}
} else if (pConn->killed) {
mError("user:%s, conn:%u is already killed", connInfo.user, pConn->id);
mndReleaseConn(pMnode, pConn);
terrno = TSDB_CODE_MND_INVALID_CONNECTION;
return -1;
}
SQueryHbRspBasic *rspBasic = taosMemoryCalloc(1, sizeof(SQueryHbRspBasic));
if (rspBasic == NULL) {
mndReleaseConn(pMnode, pConn);
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("user:%s, conn:%u failed to process hb while since %s", pConn->user, pBasic->connId, terrstr());
return -1;
}
mndSaveQueryList(pConn, pBasic);
if (pConn->killed != 0) {
rspBasic->killConnection = 1;
}
if (pConn->killId != 0) {
rspBasic->killRid = pConn->killId;
pConn->killId = 0;
}
rspBasic->connId = pConn->id;
rspBasic->totalDnodes = 1; //TODO
rspBasic->onlineDnodes = 1; //TODO
mndGetMnodeEpSet(pMnode, &rspBasic->epSet);
mndReleaseConn(pMnode, pConn);
hbRsp.query = rspBasic;
}
int32_t kvNum = taosHashGetSize(pHbReq->info);
if (NULL == pHbReq->info || kvNum <= 0) {
taosArrayPush(pBatchRsp->rsps, &hbRsp);
return TSDB_CODE_SUCCESS;
}
hbRsp.info = taosArrayInit(kvNum, sizeof(SKv));
if (NULL == hbRsp.info) {
mError("taosArrayInit %d rsp kv failed", kvNum);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
void *pIter = taosHashIterate(pHbReq->info, NULL);
while (pIter != NULL) {
SKv *kv = pIter;
switch (kv->key) {
case HEARTBEAT_KEY_DBINFO: {
void *rspMsg = NULL;
int32_t rspLen = 0;
mndValidateDbInfo(pMnode, kv->value, kv->valueLen / sizeof(SDbVgVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) {
SKv kv1 = {.key = HEARTBEAT_KEY_DBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1);
}
break;
}
case HEARTBEAT_KEY_STBINFO: {
void *rspMsg = NULL;
int32_t rspLen = 0;
mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableMetaVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) {
SKv kv1 = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1);
}
break;
}
default:
mError("invalid kv key:%d", kv->key);
hbRsp.status = TSDB_CODE_MND_APP_ERROR;
break;
}
pIter = taosHashIterate(pHbReq->info, pIter);
}
taosArrayPush(pBatchRsp->rsps, &hbRsp);
return TSDB_CODE_SUCCESS;
}
static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode;
......@@ -345,50 +444,9 @@ static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq) {
int32_t sz = taosArrayGetSize(batchReq.reqs);
for (int i = 0; i < sz; i++) {
SClientHbReq *pHbReq = taosArrayGet(batchReq.reqs, i);
if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_QUERY) {
int32_t kvNum = taosHashGetSize(pHbReq->info);
if (NULL == pHbReq->info || kvNum <= 0) {
continue;
}
SClientHbRsp hbRsp = {.connKey = pHbReq->connKey, .status = 0, .info = taosArrayInit(kvNum, sizeof(SKv))};
void *pIter = taosHashIterate(pHbReq->info, NULL);
while (pIter != NULL) {
SKv *kv = pIter;
switch (kv->key) {
case HEARTBEAT_KEY_DBINFO: {
void *rspMsg = NULL;
int32_t rspLen = 0;
mndValidateDbInfo(pMnode, kv->value, kv->valueLen / sizeof(SDbVgVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) {
SKv kv1 = {.key = HEARTBEAT_KEY_DBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1);
}
break;
}
case HEARTBEAT_KEY_STBINFO: {
void *rspMsg = NULL;
int32_t rspLen = 0;
mndValidateStbInfo(pMnode, kv->value, kv->valueLen / sizeof(SSTableMetaVersion), &rspMsg, &rspLen);
if (rspMsg && rspLen > 0) {
SKv kv1 = {.key = HEARTBEAT_KEY_STBINFO, .valueLen = rspLen, .value = rspMsg};
taosArrayPush(hbRsp.info, &kv1);
}
break;
}
default:
mError("invalid kv key:%d", kv->key);
hbRsp.status = TSDB_CODE_MND_APP_ERROR;
break;
}
pIter = taosHashIterate(pHbReq->info, pIter);
}
taosArrayPush(batchRsp.rsps, &hbRsp);
} else if (pHbReq->connKey.hbType == HEARTBEAT_TYPE_MQ) {
if (pHbReq->connKey.connType == CONN_TYPE__QUERY) {
mndProcessQueryHeartBeat(pMnode, &pReq->rpcMsg, pHbReq, &batchRsp);
} else if (pHbReq->connKey.connType == CONN_TYPE__TMQ) {
SClientHbRsp *pRsp = mndMqHbBuildRsp(pMnode, pHbReq);
if (pRsp != NULL) {
taosArrayPush(batchRsp.rsps, pRsp);
......@@ -416,73 +474,8 @@ static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq) {
taosArrayDestroy(batchRsp.rsps);
pReq->rspLen = tlen;
pReq->pRsp = buf;
return 0;
#if 0
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SHeartBeatReq *pHeartbeat = pReq->rpcMsg.pCont;
pHeartbeat->connId = htonl(pHeartbeat->connId);
pHeartbeat->pid = htonl(pHeartbeat->pid);
SConnObj *pConn = mndAcquireConn(pMnode, pHeartbeat->connId);
if (pConn == NULL) {
pConn = mndCreateConn(pMnode, &info, pHeartbeat->pid, pHeartbeat->app, 0);
if (pConn == NULL) {
mError("user:%s, conn:%d is freed and failed to create new since %s", pReq->user, pHeartbeat->connId, terrstr());
return -1;
} else {
mDebug("user:%s, conn:%d is freed and create a new conn:%d", pReq->user, pHeartbeat->connId, pConn->id);
}
} else if (pConn->killed) {
mError("user:%s, conn:%d is already killed", pReq->user, pConn->id);
terrno = TSDB_CODE_MND_INVALID_CONNECTION;
return -1;
} else {
if (pConn->ip != info.clientIp || pConn->port != info.clientPort /* || strcmp(pConn->user, info.user) != 0 */) {
char oldIpStr[40];
char newIpStr[40];
taosIpPort2String(pConn->ip, pConn->port, oldIpStr);
taosIpPort2String(info.clientIp, info.clientPort, newIpStr);
mError("conn:%d, incoming conn user:%s ip:%s, not match exist user:%s ip:%s", pConn->id, info.user, newIpStr,
pConn->user, oldIpStr);
if (pMgmt->connId < pConn->id) pMgmt->connId = pConn->id + 1;
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
terrno = TSDB_CODE_MND_INVALID_CONNECTION;
return -1;
}
}
SHeartBeatRsp *pRsp = rpcMallocCont(sizeof(SHeartBeatRsp));
if (pRsp == NULL) {
mndReleaseConn(pMnode, pConn);
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("user:%s, conn:%d failed to process hb while since %s", pReq->user, pHeartbeat->connId, terrstr());
return -1;
}
mndSaveQueryStreamList(pConn, pHeartbeat);
if (pConn->killed != 0) {
pRsp->killConnection = 1;
}
if (pConn->queryId != 0) {
pRsp->queryId = htonl(pConn->queryId);
pConn->queryId = 0;
}
pRsp->connId = htonl(pConn->id);
pRsp->totalDnodes = htonl(1);
pRsp->onlineDnodes = htonl(1);
mndGetMnodeEpSet(pMnode, &pRsp->epSet);
mndReleaseConn(pMnode, pConn);
pReq->contLen = sizeof(SConnectRsp);
pReq->pRsp = pRsp;
return 0;
#endif
}
static int32_t mndProcessKillQueryReq(SNodeMsg *pReq) {
......@@ -513,7 +506,7 @@ static int32_t mndProcessKillQueryReq(SNodeMsg *pReq) {
return -1;
} else {
mInfo("connId:%d, queryId:%d is killed by user:%s", killReq.connId, killReq.queryId, pReq->user);
pConn->queryId = killReq.queryId;
pConn->killId = killReq.queryId;
taosCacheRelease(pMgmt->cache, (void **)&pConn, false);
return 0;
}
......@@ -571,7 +564,7 @@ static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int
cols = 0;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int32_t *)pWrite = pConn->id;
*(uint32_t *)pWrite = pConn->id;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
......@@ -613,6 +606,7 @@ static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int
static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode;
int32_t numOfRows = 0;
#if 0
SConnObj *pConn = NULL;
int32_t cols = 0;
char *pWrite;
......@@ -709,6 +703,7 @@ static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, i
}
pShow->numOfRows += numOfRows;
#endif
return numOfRows;
}
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ 73a3f9bb
......@@ -18,6 +18,7 @@
#include "mndDb.h"
#include "mndDnode.h"
#include "mndInfoSchema.h"
#include "mndPerfSchema.h"
#include "mndMnode.h"
#include "mndShow.h"
#include "mndTrans.h"
......@@ -1516,6 +1517,11 @@ static int32_t mndProcessTableMetaReq(SNodeMsg *pReq) {
if (mndBuildInsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp) != 0) {
goto RETRIEVE_META_OVER;
}
} else if (0 == strcmp(infoReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {
mDebug("performance_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
if (mndBuildPerfsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp) != 0) {
goto RETRIEVE_META_OVER;
}
} else {
mDebug("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
if (mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp) != 0) {
......
source/dnode/mnode/impl/src/mnode.c
浏览文件 @ 73a3f9bb
......@@ -23,6 +23,7 @@
#include "mndDnode.h"
#include "mndFunc.h"
#include "mndInfoSchema.h"
#include "mndPerfSchema.h"
#include "mndMnode.h"
#include "mndOffset.h"
#include "mndProfile.h"
......@@ -210,6 +211,7 @@ static int32_t mndInitSteps(SMnode *pMnode, bool deploy) {
if (mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-stb", mndInitSma, mndCleanupSma) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-infos", mndInitInfos, mndCleanupInfos) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-perfs", mndInitPerfs, mndCleanupPerfs) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
if (mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc) != 0) return -1;
if (deploy) {
......
source/dnode/mnode/impl/test/profile/profile.cpp
浏览文件 @ 73a3f9bb
......@@ -82,7 +82,7 @@ TEST_F(MndTestProfile, 04_HeartBeatMsg) {
SClientHbBatchReq batchReq = {0};
batchReq.reqs = taosArrayInit(0, sizeof(SClientHbReq));
SClientHbReq req = {0};
req.connKey = {.connId = 123, .hbType = HEARTBEAT_TYPE_MQ};
req.connKey = {.connId = 123, .hbType = CONN_TYPE__TMQ};
req.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
SKv kv = {0};
kv.key = 123;
......
source/dnode/vnode/CMakeLists.txt
浏览文件 @ 73a3f9bb
......@@ -4,13 +4,13 @@ target_sources(
vnode
PRIVATE
# vnode
"src/vnd/vnodeOpen.c"
"src/vnd/vnodeArenaMAImpl.c"
"src/vnd/vnodeBufferPool.c"
# "src/vnd/vnodeBufferPool2.c"
"src/vnd/vnodeCfg.c"
"src/vnd/vnodeCommit.c"
"src/vnd/vnodeInt.c"
"src/vnd/vnodeMain.c"
"src/vnd/vnodeQuery.c"
"src/vnd/vnodeStateMgr.c"
"src/vnd/vnodeWrite.c"
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 73a3f9bb
......@@ -42,11 +42,12 @@ typedef struct STsdbCfg STsdbCfg; // todo: remove
typedef struct STqCfg STqCfg; // todo: remove
typedef struct SVnodeCfg SVnodeCfg;
int vnodeInit();
int vnodeInit(int nthreads);
void vnodeCleanup();
int vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs);
void vnodeDestroy(const char *path);
SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg);
void vnodeClose(SVnode *pVnode);
void vnodeDestroy(const char *path);
void vnodePreprocessWriteReqs(SVnode *pVnode, SArray *pMsgs);
int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
int vnodeProcessCMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
......
source/dnode/vnode/src/inc/vnd.h
浏览文件 @ 73a3f9bb
......@@ -30,6 +30,8 @@ extern "C" {
#define vTrace(...) do { if (vDebugFlag & DEBUG_TRACE) { taosPrintLog("VND ", DEBUG_TRACE, vDebugFlag, __VA_ARGS__); }} while(0)
// clang-format on
// vnodeCfg ====================
// vnodeModule ====================
int vnodeScheduleTask(int (*execute)(void*), void* arg);
......@@ -38,6 +40,10 @@ int vnodeQueryOpen(SVnode* pVnode);
void vnodeQueryClose(SVnode* pVnode);
int vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg);
// vnodeCommit ====================
int vnodeSaveInfo(const char* dir, const SVnodeInfo* pCfg);
int vnodeCommitInfo(const char* dir, const SVnodeInfo* pInfo);
#if 1
// SVBufPool
int vnodeOpenBufPool(SVnode* pVnode);
......@@ -75,9 +81,9 @@ void vmaFree(SVMemAllocator* pVMA, void* ptr);
bool vmaIsFull(SVMemAllocator* pVMA);
// vnodeCfg.h
extern const SVnodeCfg defaultVnodeOptions;
extern const SVnodeCfg vnodeCfgDefault;
int vnodeValidateOptions(const SVnodeCfg*);
int vnodeCheckCfg(const SVnodeCfg*);
void vnodeOptionsCopy(SVnodeCfg* pDest, const SVnodeCfg* pSrc);
// For commit
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 73a3f9bb
......@@ -43,6 +43,7 @@
extern "C" {
#endif
typedef struct SVnodeInfo SVnodeInfo;
typedef struct SMeta SMeta;
typedef struct STsdb STsdb;
typedef struct STQ STQ;
......@@ -72,6 +73,11 @@ struct SVState {
int64_t applied;
};
struct SVnodeInfo {
SVnodeCfg config;
SVState state;
};
struct SVnode {
int32_t vgId;
char* path;
......
source/dnode/vnode/src/vnd/vnodeCfg.c
浏览文件 @ 73a3f9bb
......@@ -15,14 +15,15 @@
#include "vnodeInt.h"
const SVnodeCfg defaultVnodeOptions = {
.wsize = 96 * 1024 * 1024, .ssize = 1 * 1024 * 1024, .lsize = 1024, .walCfg = {.level = TAOS_WAL_WRITE}}; /* TODO */
const SVnodeCfg vnodeCfgDefault = {
.wsize = 96 * 1024 * 1024, .ssize = 1 * 1024 * 1024, .lsize = 1024, .walCfg = {.level = TAOS_WAL_WRITE}};
int vnodeValidateOptions(const SVnodeCfg *pVnodeOptions) {
int vnodeCheckCfg(const SVnodeCfg *pCfg) {
// TODO
return 0;
}
#if 1 //======================================================================
void vnodeOptionsCopy(SVnodeCfg *pDest, const SVnodeCfg *pSrc) {
memcpy((void *)pDest, (void *)pSrc, sizeof(SVnodeCfg));
}
......@@ -46,3 +47,5 @@ int vnodeValidateTableHash(SVnodeCfg *pVnodeOptions, char *tableFName) {
return TSDB_CODE_SUCCESS;
}
#endif
\ No newline at end of file
source/dnode/vnode/src/vnd/vnodeCommit.c
浏览文件 @ 73a3f9bb
......@@ -15,11 +15,86 @@
#include "vnodeInt.h"
#define VND_INFO_FNAME "vnode.json"
#define VND_INFO_FNAME_TMP "vnode_tmp.json"
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, uint8_t **ppData, int *len);
static int vnodeDecodeInfo(uint8_t *pData, int len, SVnodeInfo *pInfo);
static int vnodeStartCommit(SVnode *pVnode);
static int vnodeEndCommit(SVnode *pVnode);
static int vnodeCommit(void *arg);
static void vnodeWaitCommit(SVnode *pVnode);
int vnodeSaveInfo(const char *dir, const SVnodeInfo *pInfo) {
char fname[TSDB_FILENAME_LEN];
TdFilePtr pFile;
uint8_t *data;
int len;
snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);
// encode info
data = NULL;
len = 0;
if (vnodeEncodeInfo(pInfo, &data, &len) < 0) {
return -1;
}
// save info to a vnode_tmp.json
pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
if (pFile == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (taosWriteFile(pFile, data, len) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
if (taosFsyncFile(pFile) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
goto _err;
}
taosCloseFile(&pFile);
// free info binary
taosMemoryFree(data);
vInfo("vgId: %d vnode info is saved, fname: %s", pInfo->config.vgId, fname);
return 0;
_err:
taosCloseFile(&pFile);
taosMemoryFree(data);
return -1;
}
int vnodeCommitInfo(const char *dir, const SVnodeInfo *pInfo) {
char fname[TSDB_FILENAME_LEN];
char tfname[TSDB_FILENAME_LEN];
snprintf(fname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME);
snprintf(tfname, TSDB_FILENAME_LEN, "%s%s%s", dir, TD_DIRSEP, VND_INFO_FNAME_TMP);
if (taosRenameFile(tfname, fname) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
vInfo("vgId: %d vnode info is committed", pInfo->config.vgId);
return 0;
}
int vnodeLoadInfo(const char *dir) {
// TODO
return 0;
}
int vnodeAsyncCommit(SVnode *pVnode) {
vnodeWaitCommit(pVnode);
......@@ -60,4 +135,14 @@ static int vnodeEndCommit(SVnode *pVnode) {
return 0;
}
static FORCE_INLINE void vnodeWaitCommit(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); }
\ No newline at end of file
static FORCE_INLINE void vnodeWaitCommit(SVnode *pVnode) { tsem_wait(&pVnode->canCommit); }
static int vnodeEncodeInfo(const SVnodeInfo *pInfo, uint8_t **ppData, int *len) {
// TODO
return 0;
}
static int vnodeDecodeInfo(uint8_t *pData, int len, SVnodeInfo *pInfo) {
// TODO
return 0;
}
source/dnode/vnode/src/vnd/vnodeMain.c → source/dnode/vnode/src/vnd/vnodeOpen.c
浏览文件 @ 73a3f9bb
......@@ -20,11 +20,40 @@ static void vnodeFree(SVnode *pVnode);
static int vnodeOpenImpl(SVnode *pVnode);
static void vnodeCloseImpl(SVnode *pVnode);
int vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
SVnodeInfo info = {0};
char dir[TSDB_FILENAME_LEN];
// TODO: check if directory exists
// check config
if (vnodeCheckCfg(pCfg) < 0) {
vError("vgId: %d failed to create vnode since: %s", pCfg->vgId, tstrerror(terrno));
return -1;
}
// create vnode env
if (tfsMkdir(pTfs, path) < 0) {
vError("vgId: %d failed to create vnode since: %s", pCfg->vgId, tstrerror(terrno));
return -1;
}
snprintf(dir, TSDB_FILENAME_LEN, "%s%s%s", tfsGetPrimaryPath(pTfs), TD_DIRSEP, path);
info.config = *pCfg;
if (vnodeSaveInfo(dir, &info) < 0 || vnodeCommitInfo(dir, &info) < 0) {
vError("vgId: %d failed to save vnode config since %s", pCfg->vgId, tstrerror(terrno));
return -1;
}
return 0;
}
SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
SVnode *pVnode = NULL;
// Set default options
SVnodeCfg cfg = defaultVnodeOptions;
SVnodeCfg cfg = vnodeCfgDefault;
if (pVnodeCfg != NULL) {
cfg.vgId = pVnodeCfg->vgId;
cfg.msgCb = pVnodeCfg->msgCb;
......@@ -36,7 +65,7 @@ SVnode *vnodeOpen(const char *path, const SVnodeCfg *pVnodeCfg) {
}
// Validate options
if (vnodeValidateOptions(&cfg) < 0) {
if (vnodeCheckCfg(&cfg) < 0) {
// TODO
return NULL;
}
......
source/libs/catalog/inc/catalogInt.h
浏览文件 @ 73a3f9bb
......@@ -227,20 +227,20 @@ typedef struct SCtgAction {
#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_SYS_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_SYS_DB(_flag) ((_flag) & CTG_FLAG_SYS_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_SYS_DB(_flag) ((_flag) |= CTG_FLAG_SYS_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)))
#define CTG_IS_SYS_DBNAME(_dbname) (((*(_dbname) == 'i') && (0 == strcmp(_dbname, TSDB_INFORMATION_SCHEMA_DB))) || ((*(_dbname) == 'p') && (0 == strcmp(_dbname, TSDB_PERFORMANCE_SCHEMA_DB))))
#define CTG_META_SIZE(pMeta) (sizeof(STableMeta) + ((pMeta)->tableInfo.numOfTags + (pMeta)->tableInfo.numOfColumns) * sizeof(SSchema))
......
source/libs/catalog/src/catalog.c
浏览文件 @ 73a3f9bb
......@@ -217,7 +217,7 @@ int32_t ctgPushRmDBMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId)
}
char *p = strchr(dbFName, '.');
if (p && CTG_IS_INF_DBNAME(p + 1)) {
if (p && CTG_IS_SYS_DBNAME(p + 1)) {
dbFName = p + 1;
}
......@@ -304,7 +304,7 @@ int32_t ctgPushUpdateVgMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t d
}
char *p = strchr(dbFName, '.');
if (p && CTG_IS_INF_DBNAME(p + 1)) {
if (p && CTG_IS_SYS_DBNAME(p + 1)) {
dbFName = p + 1;
}
......@@ -336,7 +336,7 @@ int32_t ctgPushUpdateTblMsgInQueue(SCatalog* pCtg, STableMetaOutput *output, boo
}
char *p = strchr(output->dbFName, '.');
if (p && CTG_IS_INF_DBNAME(p + 1)) {
if (p && CTG_IS_SYS_DBNAME(p + 1)) {
memmove(output->dbFName, p + 1, strlen(p + 1));
}
......@@ -410,7 +410,7 @@ void ctgWReleaseVgInfo(SCtgDBCache *dbCache) {
int32_t ctgAcquireDBCacheImpl(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache, bool acquire) {
char *p = strchr(dbFName, '.');
if (p && CTG_IS_INF_DBNAME(p + 1)) {
if (p && CTG_IS_SYS_DBNAME(p + 1)) {
dbFName = p + 1;
}
......@@ -688,7 +688,7 @@ int32_t ctgGetTableMetaFromCache(SCatalog* pCtg, const SName* pTableName, STable
}
char dbFName[TSDB_DB_FNAME_LEN] = {0};
if (CTG_FLAG_IS_INF_DB(flag)) {
if (CTG_FLAG_IS_SYS_DB(flag)) {
strcpy(dbFName, pTableName->dbname);
} else {
tNameGetFullDbName(pTableName, dbFName);
......@@ -1721,7 +1721,7 @@ int32_t ctgRefreshTblMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps,
SVgroupInfo vgroupInfo = {0};
int32_t code = 0;
if (!CTG_FLAG_IS_INF_DB(flag)) {
if (!CTG_FLAG_IS_SYS_DB(flag)) {
CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
}
......@@ -1732,7 +1732,7 @@ int32_t ctgRefreshTblMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps,
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
}
if (CTG_FLAG_IS_INF_DB(flag)) {
if (CTG_FLAG_IS_SYS_DB(flag)) {
ctgDebug("will refresh tbmeta, supposed in information_schema, tbName:%s", tNameGetTableName(pTableName));
CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, (char *)pTableName->dbname, (char *)pTableName->tname, output));
......@@ -1820,8 +1820,8 @@ int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, cons
uint64_t suid = 0;
STableMetaOutput *output = NULL;
if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
CTG_FLAG_SET_INF_DB(flag);
if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
CTG_FLAG_SET_SYS_DB(flag);
}
CTG_ERR_RET(ctgGetTableMetaFromCache(pCtg, pTableName, pTableMeta, &inCache, flag, &dbId));
......@@ -1829,7 +1829,7 @@ int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, cons
int32_t tbType = 0;
if (inCache) {
if (CTG_FLAG_MATCH_STB(flag, (*pTableMeta)->tableType) && ((!CTG_FLAG_IS_FORCE_UPDATE(flag)) || (CTG_FLAG_IS_INF_DB(flag)))) {
if (CTG_FLAG_MATCH_STB(flag, (*pTableMeta)->tableType) && ((!CTG_FLAG_IS_FORCE_UPDATE(flag)) || (CTG_FLAG_IS_SYS_DB(flag)))) {
goto _return;
}
......@@ -1885,7 +1885,7 @@ _return:
if (CTG_TABLE_NOT_EXIST(code) && inCache) {
char dbFName[TSDB_DB_FNAME_LEN] = {0};
if (CTG_FLAG_IS_INF_DB(flag)) {
if (CTG_FLAG_IS_SYS_DB(flag)) {
strcpy(dbFName, pTableName->dbname);
} else {
tNameGetFullDbName(pTableName, dbFName);
......@@ -2633,7 +2633,7 @@ int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgm
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
......@@ -2666,7 +2666,7 @@ _return:
int32_t catalogGetTableHashVgroup(SCatalog *pCtg, void *pTrans, const SEpSet *pMgmtEps, const SName *pTableName, SVgroupInfo *pVgroup) {
CTG_API_ENTER();
if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
......
source/libs/function/src/builtins.c
浏览文件 @ 73a3f9bb
......@@ -782,6 +782,26 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.sprocessFunc = timeDiffFunction,
.finalizeFunc = NULL
},
{
.name = "now",
.type = FUNCTION_TYPE_NOW,
.classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
.translateFunc = translateTimePseudoColumn,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = nowFunction,
.finalizeFunc = NULL
},
{
.name = "today",
.type = FUNCTION_TYPE_TODAY,
.classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
.translateFunc = translateTimePseudoColumn,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = todayFunction,
.finalizeFunc = NULL
},
{
.name = "_rowts",
.type = FUNCTION_TYPE_ROWTS,
......@@ -851,16 +871,6 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.initFunc = NULL,
.sprocessFunc = winDurFunction,
.finalizeFunc = NULL
},
{
.name = "now",
.type = FUNCTION_TYPE_NOW,
.classification = FUNC_MGT_SCALAR_FUNC | FUNC_MGT_DATETIME_FUNC,
.translateFunc = translateTimePseudoColumn,
.getEnvFunc = getTimePseudoFuncEnv,
.initFunc = NULL,
.sprocessFunc = winDurFunction,
.finalizeFunc = NULL
}
};
......
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @ 73a3f9bb
......@@ -1201,7 +1201,7 @@ void valueNodeToVariant(const SValueNode* pNode, SVariant* pVal) {
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
pVal->pz = pNode->datum.p;
pVal->pz = pNode->datum.p + VARSTR_HEADER_SIZE;
break;
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_DECIMAL:
......
source/libs/parser/inc/parAst.h
浏览文件 @ 73a3f9bb
......@@ -92,6 +92,7 @@ SNode* createOperatorNode(SAstCreateContext* pCxt, EOperatorType type, SNode* pL
SNode* createBetweenAnd(SAstCreateContext* pCxt, SNode* pExpr, SNode* pLeft, SNode* pRight);
SNode* createNotBetweenAnd(SAstCreateContext* pCxt, SNode* pExpr, SNode* pLeft, SNode* pRight);
SNode* createFunctionNode(SAstCreateContext* pCxt, const SToken* pFuncName, SNodeList* pParameterList);
SNode* createFunctionNodeNoParam(SAstCreateContext* pCxt, const SToken* pFuncName);
SNode* createCastFunctionNode(SAstCreateContext* pCxt, SNode* pExpr, SDataType dt);
SNode* createNodeListNode(SAstCreateContext* pCxt, SNodeList* pList);
SNode* createNodeListNodeEx(SAstCreateContext* pCxt, SNode* p1, SNode* p2);
......
source/libs/parser/inc/sql.y
浏览文件 @ 73a3f9bb
......@@ -503,6 +503,8 @@ column_name(A) ::= NK_ID(B).
function_name(A) ::= NK_ID(B). { A = B; }
function_name(A) ::= FIRST(B). { A = B; }
function_name(A) ::= LAST(B). { A = B; }
function_name(A) ::= NOW(B). { A = B; }
function_name(A) ::= TODAY(B). { A = B; }
%type table_alias { SToken }
%destructor table_alias { }
......@@ -535,6 +537,7 @@ expression(A) ::= pseudo_column(B).
expression(A) ::= column_reference(B). { A = B; }
expression(A) ::= function_name(B) NK_LP expression_list(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, C)); }
expression(A) ::= function_name(B) NK_LP NK_STAR(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, createNodeList(pCxt, createColumnNode(pCxt, NULL, &C)))); }
expression(A) ::= function_name(B) NK_LP NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNodeNoParam(pCxt, &B)); }
expression(A) ::= CAST(B) NK_LP expression(C) AS type_name(D) NK_RP(E). { A = createRawExprNodeExt(pCxt, &B, &E, createCastFunctionNode(pCxt, releaseRawExprNode(pCxt, C), D)); }
//expression(A) ::= case_expression(B). { A = B; }
expression(A) ::= subquery(B). { A = B; }
......@@ -581,8 +584,8 @@ expression_list(A) ::= expression_list(B) NK_COMMA expression(C).
column_reference(A) ::= column_name(B). { A = createRawExprNode(pCxt, &B, createColumnNode(pCxt, NULL, &B)); }
column_reference(A) ::= table_name(B) NK_DOT column_name(C). { A = createRawExprNodeExt(pCxt, &B, &C, createColumnNode(pCxt, &B, &C)); }
pseudo_column(A) ::= NOW(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= TODAY(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
//pseudo_column(A) ::= NOW(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
//pseudo_column(A) ::= TODAY(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= ROWTS(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= TBNAME(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= QSTARTTS(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
......
source/libs/parser/src/parAstCreater.c
浏览文件 @ 73a3f9bb
......@@ -360,6 +360,39 @@ SNode* createFunctionNode(SAstCreateContext* pCxt, const SToken* pFuncName, SNod
return (SNode*)func;
}
SNode* createFunctionNodeNoParam(SAstCreateContext* pCxt, const SToken* pFuncName) {
SFunctionNode* func = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
CHECK_OUT_OF_MEM(func);
char buf[64] = {0};
int32_t dataType;
switch (pFuncName->type) {
case TK_NOW: {
int64_t ts = taosGetTimestamp(TSDB_TIME_PRECISION_MILLI);
snprintf(buf, sizeof(buf), "%"PRId64, ts);
dataType = TSDB_DATA_TYPE_BIGINT;
break;
}
case TK_TODAY: {
int64_t ts = taosGetTimestampToday(TSDB_TIME_PRECISION_MILLI);
snprintf(buf, sizeof(buf), "%"PRId64, ts);
dataType = TSDB_DATA_TYPE_BIGINT;
break;
}
//case TK_TIMEZONE: {
// strncpy(buf, tsTimezoneStr, strlen(tsTimezoneStr));
// dataType = TSDB_DATA_TYPE_BINARY;
// break;
//}
}
SToken token = {.type = pFuncName->type, .n = strlen(buf), .z = buf};
SNodeList *pParameterList = createNodeList(pCxt, createValueNode(pCxt, dataType, &token));
strncpy(func->functionName, pFuncName->z, pFuncName->n);
func->pParameterList = pParameterList;
return (SNode*)func;
}
SNode* createCastFunctionNode(SAstCreateContext* pCxt, SNode* pExpr, SDataType dt) {
SFunctionNode* func = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
CHECK_OUT_OF_MEM(func);
......
source/libs/parser/src/parInsert.c
浏览文件 @ 73a3f9bb
......@@ -223,11 +223,15 @@ static int32_t createSName(SName* pName, SToken* pTableName, SParseContext* pPar
return code;
}
static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SToken* pTname, bool isStb) {
SParseContext* pBasicCtx = pCxt->pComCxt;
SName name = {0};
createSName(&name, pTname, pBasicCtx, &pCxt->msg);
CHECK_CODE(catalogGetTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name, &pCxt->pTableMeta));
if (isStb) {
CHECK_CODE(catalogGetSTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name, &pCxt->pTableMeta));
} else {
CHECK_CODE(catalogGetTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name, &pCxt->pTableMeta));
}
SVgroupInfo vg;
CHECK_CODE(catalogGetTableHashVgroup(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name, &vg));
CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));
......@@ -235,6 +239,15 @@ static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
return TSDB_CODE_SUCCESS;
}
static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
return getTableMetaImpl(pCxt, pTname, false);
}
static int32_t getSTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
return getTableMetaImpl(pCxt, pTname, true);
}
static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pSchema) {
while (start < end) {
if (strlen(pSchema[start].name) == pColname->n && strncmp(pColname->z, pSchema[start].name, pColname->n) == 0) {
......@@ -818,7 +831,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken)
SToken sToken;
// pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
NEXT_TOKEN(pCxt->pSql, sToken);
CHECK_CODE(getTableMeta(pCxt, &sToken));
CHECK_CODE(getSTableMeta(pCxt, &sToken));
if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
}
......
source/libs/parser/src/sql.c
浏览文件 @ 73a3f9bb
此差异已折叠。
source/libs/scalar/src/sclfunc.c
浏览文件 @ 73a3f9bb
......@@ -1259,6 +1259,22 @@ int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *p
return TSDB_CODE_SUCCESS;
}
int32_t nowFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
if (inputNum != 1) {
return TSDB_CODE_FAILED;
}
colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 0));
return TSDB_CODE_SUCCESS;
}
int32_t todayFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
if (inputNum != 1) {
return TSDB_CODE_FAILED;
}
colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 0));
return TSDB_CODE_SUCCESS;
}
int32_t atanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
return doScalarFunctionUnique(pInput, inputNum, pOutput, atan);
}
......
source/libs/scheduler/inc/schedulerInt.h
浏览文件 @ 73a3f9bb
......@@ -156,8 +156,8 @@ typedef struct SSchJob {
int32_t levelNum;
int32_t taskNum;
void *transport;
SArray *nodeList; // qnode/vnode list, element is SQueryNodeAddr
SArray *levels; // Element is SQueryLevel, starting from 0. SArray<SSchLevel>
SArray *nodeList; // qnode/vnode list, SArray<SQueryNodeAddr>
SArray *levels; // starting from 0. SArray<SSchLevel>
SNodeList *subPlans; // subplan pointer copied from DAG, no need to free it in scheduler
int32_t levelIdx;
......
source/libs/scheduler/src/scheduler.c
浏览文件 @ 73a3f9bb
......@@ -2655,6 +2655,34 @@ _return:
SCH_RET(code);
}
int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub) {
int32_t code = 0;
SSchJob *pJob = schAcquireJob(job);
if (NULL == pJob) {
qDebug("acquire job from jobRef list failed, may not started or dropped, refId:%" PRIx64, job);
SCH_ERR_RET(TSDB_CODE_SCH_STATUS_ERROR);
}
if (pJob->status < JOB_TASK_STATUS_NOT_START || pJob->levelNum <= 0 || NULL == pJob->levels) {
qDebug("job not initialized or not executable job, refId:%" PRIx64, job);
SCH_ERR_RET(TSDB_CODE_SCH_STATUS_ERROR);
}
for (int32_t i = pJob->levelNum - 1; i >= 0; --i) {
SSchLevel *pLevel = taosArrayGet(pJob->levels, i);
for (int32_t m = 0; m < pLevel->taskNum; ++m) {
SSchTask *pTask = taosArrayGet(pLevel->subTasks, m);
SQuerySubDesc subDesc = {.tid = pTask->taskId, .status = pTask->status};
taosArrayPush(pSub, &subDesc);
}
}
return TSDB_CODE_SUCCESS;
}
int32_t scheduleCancelJob(int64_t job) {
SSchJob *pJob = schAcquireJob(job);
if (NULL == pJob) {
......@@ -2672,7 +2700,7 @@ int32_t scheduleCancelJob(int64_t job) {
void schedulerFreeJob(int64_t job) {
SSchJob *pJob = schAcquireJob(job);
if (NULL == pJob) {
qError("acquire job from jobRef list failed, may be dropped, refId:%" PRIx64, job);
qDebug("acquire job from jobRef list failed, may be dropped, refId:%" PRIx64, job);
return;
}
......
source/libs/transport/src/transComm.c
浏览文件 @ 73a3f9bb
......@@ -292,7 +292,7 @@ void* transCtxDumpBrokenlinkVal(STransCtx* ctx, int32_t* msgType) {
void transQueueInit(STransQueue* queue, void (*freeFunc)(const void* arg)) {
queue->q = taosArrayInit(2, sizeof(void*));
queue->freeFunc = freeFunc;
queue->freeFunc = (void (*)(const void*))freeFunc;
}
bool transQueuePush(STransQueue* queue, void* arg) {
if (queue->q == NULL) {
......
source/libs/transport/test/transportTests.cpp
浏览文件 @ 73a3f9bb
......@@ -156,14 +156,14 @@ TEST_F(TransCtxEnv, mergeTest) {
STransCtx *src = (STransCtx *)taosMemoryCalloc(1, sizeof(STransCtx));
transCtxInit(src);
{
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = (void (*)(const void*))taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
}
{
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = (void (*)(const void*))taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
......@@ -176,14 +176,14 @@ TEST_F(TransCtxEnv, mergeTest) {
STransCtx *src = (STransCtx *)taosMemoryCalloc(1, sizeof(STransCtx));
transCtxInit(src);
{
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = (void (*)(const void*))taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
}
{
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = (void (*)(const void*))taosMemoryFree};
val1.val = taosMemoryMalloc(12);
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
key++;
......@@ -198,7 +198,7 @@ TEST_F(TransCtxEnv, mergeTest) {
STransCtx *src = (STransCtx *)taosMemoryCalloc(1, sizeof(STransCtx));
transCtxInit(src);
{
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = (void (*)(const void*))taosMemoryFree};
val1.val = taosMemoryCalloc(1, 11);
memcpy(val1.val, val.c_str(), val.size());
......@@ -206,7 +206,7 @@ TEST_F(TransCtxEnv, mergeTest) {
key++;
}
{
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = taosMemoryFree};
STransCtxVal val1 = {.val = NULL, .clone = NULL, .freeFunc = (void (*)(const void*))taosMemoryFree};
val1.val = taosMemoryCalloc(1, 11);
memcpy(val1.val, val.c_str(), val.size());
taosHashPut(src->args, &key, sizeof(key), &val1, sizeof(val1));
......
source/os/src/osMemory.c
浏览文件 @ 73a3f9bb
......@@ -156,7 +156,7 @@ void *taosMemoryStrDup(void *ptr) {
}
void taosMemoryFree(const void *ptr) {
void taosMemoryFree(void *ptr) {
if (ptr == NULL) return;
#ifdef USE_TD_MEMORY
......@@ -166,10 +166,10 @@ void taosMemoryFree(const void *ptr) {
// memset(pTdMemoryInfo, 0, sizeof(TdMemoryInfo));
free(pTdMemoryInfo);
} else {
free((void*)ptr);
free(ptr);
}
#else
return free((void*)ptr);
return free(ptr);
#endif
}
......
source/util/src/tarray.c
浏览文件 @ 73a3f9bb
......@@ -303,6 +303,21 @@ void taosArrayClear(SArray* pArray) {
pArray->size = 0;
}
void taosArrayClearEx(SArray* pArray, void (*fp)(void*)) {
if (pArray == NULL) return;
if (fp == NULL) {
pArray->size = 0;
return;
}
for (int32_t i = 0; i < pArray->size; ++i) {
fp(TARRAY_GET_ELEM(pArray, i));
}
pArray->size = 0;
}
void* taosArrayDestroy(SArray* pArray) {
if (pArray) {
taosMemoryFree(pArray->pData);
......
tools/shell/src/shellEngine.c
浏览文件 @ 73a3f9bb
......@@ -344,6 +344,8 @@ void shellRunCommandOnServer(TAOS *con, char command[]) {
atomic_store_64(&result, 0);
freeResultWithRid(oresult);
taos_free_result(pSql);
return;
}
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或