Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1187
Star 22018
Fork 4786
T
TDengine
提交 60b56c66 编写于 作者: S Shengliang Guan
浏览文件
操作

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

上级 ecfd874c 164a80e9
展开全部 隐藏空白更改
内联 并排
Showing with 1365 addition and 1021 deletion
.devcontainer/devcontainer.json
浏览文件 @ 60b56c66
......@@ -18,7 +18,8 @@
"ms-vscode.cpptools",
"ms-vscode.cmake-tools",
"austin.code-gnu-global",
"visualstudioexptteam.vscodeintel"
"visualstudioexptteam.vscodeintel",
"eamodio.gitlens"
],
// Use 'forwardPorts' to make a list of ports inside the container available locally.
......
include/common/tmsg.h
浏览文件 @ 60b56c66
......@@ -25,7 +25,9 @@ extern "C" {
#include "taoserror.h"
#include "tcoding.h"
#include "tdataformat.h"
#include "tlist.h"
/* ------------------------ MESSAGE DEFINITIONS ------------------------ */
#define TD_MSG_NUMBER_
#undef TD_MSG_DICT_
#undef TD_MSG_INFO_
......@@ -54,6 +56,47 @@ extern int tMsgDict[];
typedef uint16_t tmsg_t;
/* ------------------------ ENCODE/DECODE FUNCTIONS AND MACROS ------------------------ */
struct SMEListNode {
TD_SLIST_NODE(SMEListNode);
SEncoder coder;
};
typedef struct SMsgEncoder {
SEncoder coder;
TD_SLIST(SMEListNode) eStack; // encode stack
} SMsgEncoder;
struct SMDFreeListNode {
TD_SLIST_NODE(SMDFreeListNode);
char payload[];
};
struct SMDListNode {
TD_SLIST_NODE(SMDListNode);
SDecoder coder;
};
typedef struct SMsgDecoder {
SDecoder coder;
TD_SLIST(SMDListNode) dStack;
TD_SLIST(SMDFreeListNode) freeList;
} SMsgDecoder;
#define TMSG_MALLOC(SIZE, DECODER) \
({ \
void* ptr = malloc((SIZE) + sizeof(struct SMDFreeListNode)); \
if (ptr) { \
TD_SLIST_PUSH(&((DECODER)->freeList), (struct SMDFreeListNode*)ptr); \
ptr = POINTER_SHIFT(ptr, sizeof(struct SMDFreeListNode*)); \
} \
ptr; \
})
void tmsgInitMsgDecoder(SMsgDecoder* pMD, td_endian_t endian, uint8_t* data, int64_t size);
void tmsgClearMsgDecoder(SMsgDecoder* pMD);
/* ------------------------ OTHER DEFINITIONS ------------------------ */
// IE type
#define TSDB_IE_TYPE_SEC 1
#define TSDB_IE_TYPE_META 2
......@@ -503,7 +546,7 @@ typedef struct {
typedef struct {
SMsgHead header;
union {
int32_t showId;
int64_t showId;
int64_t qhandle;
int64_t qId;
}; // query handle
......@@ -823,7 +866,7 @@ typedef struct {
} SCompactMsg;
typedef struct SShowRsp {
int32_t showId;
int64_t showId;
STableMetaMsg tableMeta;
} SShowRsp;
......@@ -1228,100 +1271,11 @@ typedef struct SVCreateTbReq {
};
} SVCreateTbReq;
static FORCE_INLINE int tSerializeSVCreateTbReq(void** buf, const SVCreateTbReq* pReq) {
int tlen = 0;
tlen += taosEncodeFixedU64(buf, pReq->ver);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedU32(buf, pReq->ttl);
tlen += taosEncodeFixedU32(buf, pReq->keep);
tlen += taosEncodeFixedU8(buf, pReq->type);
switch (pReq->type) {
case TD_SUPER_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nCols);
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pSchema[i].name);
}
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nTagCols);
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
tlen += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
tlen += taosEncodeFixedU32(buf, pReq->ntbCfg.nCols);
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->ntbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return tlen;
}
int tmsgSVCreateTbReqEncode(SMsgEncoder* pCoder, SVCreateTbReq* pReq);
int tmsgSVCreateTbReqDecode(SMsgDecoder* pCoder, SVCreateTbReq* pReq);
int tSerializeSVCreateTbReq(void** buf, const SVCreateTbReq* pReq);
void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
static FORCE_INLINE void* tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq) {
buf = taosDecodeFixedU64(buf, &(pReq->ver));
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type));
switch (pReq->type) {
case TD_SUPER_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
buf = taosDecodeFixedU32(buf, &(pReq->stbCfg.nCols));
pReq->stbCfg.pSchema = (SSchema*)malloc(pReq->stbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pSchema[i].type));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].colId));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].bytes));
buf = taosDecodeStringTo(buf, pReq->stbCfg.pSchema[i].name);
}
buf = taosDecodeFixedU32(buf, &pReq->stbCfg.nTagCols);
pReq->stbCfg.pTagSchema = (SSchema*)malloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
buf = taosDecodeFixedU64(buf, &pReq->ctbCfg.suid);
buf = tdDecodeKVRow(buf, &pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
buf = taosDecodeFixedU32(buf, &pReq->ntbCfg.nCols);
pReq->ntbCfg.pSchema = (SSchema*)malloc(pReq->ntbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &pReq->ntbCfg.pSchema[i].type);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return buf;
}
typedef struct SVCreateTbRsp {
} SVCreateTbRsp;
......@@ -1336,7 +1290,7 @@ typedef struct SVShowTablesRsp {
typedef struct SVShowTablesFetchReq {
SMsgHead head;
int64_t id;
int32_t id;
} SVShowTablesFetchReq;
typedef struct SVShowTablesFetchRsp {
......
include/common/tname.h
浏览文件 @ 60b56c66
......@@ -17,6 +17,7 @@
#define TDENGINE_TNAME_H
#include "tdef.h"
#include "tmsg.h"
#define TSDB_DB_NAME_T 1
#define TSDB_TABLE_NAME_T 2
......@@ -58,4 +59,6 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type);
int32_t tNameSetAcctId(SName* dst, int32_t acctId);
SSchema createSchema(uint8_t type, int32_t bytes, int32_t colId, const char* name);
#endif // TDENGINE_TNAME_H
include/dnode/vnode/meta/meta.h
浏览文件 @ 60b56c66
......@@ -37,6 +37,13 @@ typedef struct SMetaCfg {
uint64_t lruSize;
} SMetaCfg;
typedef struct {
int32_t nCols;
SSchema *pSchema;
} SSchemaWrapper;
typedef struct SMTbCursor SMTbCursor;
typedef SVCreateTbReq STbCfg;
// SMeta operations
......@@ -48,7 +55,13 @@ int metaDropTable(SMeta *pMeta, tb_uid_t uid);
int metaCommit(SMeta *pMeta);
// For Query
int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg);
STbCfg * metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid);
STbCfg * metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid);
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline);
SMTbCursor * metaOpenTbCursor(SMeta *pMeta);
void metaCloseTbCursor(SMTbCursor *pTbCur);
char *metaTbCursorNext(SMTbCursor *pTbCur);
// Options
void metaOptionsInit(SMetaCfg *pMetaCfg);
......
include/libs/catalog/catalog.h
浏览文件 @ 60b56c66
......@@ -54,11 +54,11 @@ int32_t catalogInit(SCatalogCfg *cfg);
/**
* Get a cluster's catalog handle for all later operations.
* @param clusterId (input, end with \0)
* @param clusterId
* @param catalogHandle (output, NO need to free it)
* @return error code
*/
int32_t catalogGetHandle(const char *clusterId, struct SCatalog** catalogHandle);
int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle);
int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version);
......
include/libs/parser/parsenodes.h
浏览文件 @ 60b56c66
......@@ -169,6 +169,7 @@ typedef struct SDclStmtInfo {
SEpSet epSet;
char* pMsg;
int32_t msgLen;
void* pExtension; // todo remove it soon
} SDclStmtInfo;
#ifdef __cplusplus
......
include/libs/parser/parser.h
浏览文件 @ 60b56c66
......@@ -74,7 +74,6 @@ int32_t getExprFunctionLevel(const SQueryStmtInfo* pQueryInfo);
STableMetaInfo* getMetaInfo(const SQueryStmtInfo* pQueryInfo, int32_t tableIndex);
SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
SSchema createSchema(uint8_t type, int16_t bytes, int16_t colId, const char* name);
int32_t getNewResColId();
void addIntoSourceParam(SSourceParam* pSourceParam, tExprNode* pNode, SColumn* pColumn);
......
include/libs/qworker/qworker.h
浏览文件 @ 60b56c66
......@@ -54,8 +54,11 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
void qWorkerDestroy(void **qWorkerMgmt);
int32_t qWorkerProcessShowMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
int32_t qWorkerProcessShowFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
void qWorkerDestroy(void **qWorkerMgmt);
#ifdef __cplusplus
}
......
include/util/encode.h
浏览文件 @ 60b56c66
......@@ -26,8 +26,8 @@ extern "C" {
typedef struct {
td_endian_t endian;
uint8_t* data;
int64_t size;
int64_t pos;
int32_t size;
int32_t pos;
} SEncoder, SDecoder;
#define tPut(TYPE, BUF, VAL) ((TYPE*)(BUF))[0] = (VAL)
......@@ -62,7 +62,7 @@ typedef struct {
#define TD_CHECK_CODER_CAPACITY_FAILED(CODER, EXPSIZE) (((CODER)->size - (CODER)->pos) < (EXPSIZE))
/* ------------------------ FOR ENCODER ------------------------ */
static FORCE_INLINE void tInitEncoder(SEncoder* pEncoder, td_endian_t endian, uint8_t* data, int64_t size) {
static FORCE_INLINE void tInitEncoder(SEncoder* pEncoder, td_endian_t endian, uint8_t* data, int32_t size) {
pEncoder->endian = endian;
pEncoder->data = data;
pEncoder->size = (data) ? size : 0;
......@@ -266,7 +266,7 @@ static FORCE_INLINE int tEncodeCStr(SEncoder* pEncoder, const char* val) {
}
/* ------------------------ FOR DECODER ------------------------ */
static FORCE_INLINE void tInitDecoder(SDecoder* pDecoder, td_endian_t endian, uint8_t* data, int64_t size) {
static FORCE_INLINE void tInitDecoder(SDecoder* pDecoder, td_endian_t endian, uint8_t* data, int32_t size) {
ASSERT(!TD_IS_NULL(data));
pDecoder->endian = endian;
pDecoder->data = data;
......
source/client/inc/clientInt.h
浏览文件 @ 60b56c66
......@@ -101,10 +101,17 @@ typedef struct SReqResultInfo {
uint32_t current;
} SReqResultInfo;
typedef struct SShowReqInfo {
int64_t execId; // showId/queryId
int32_t vgId;
SArray *pArray; // SArray<SVgroupInfo>
int32_t currentIndex; // current accessed vgroup index.
} SShowReqInfo;
typedef struct SRequestSendRecvBody {
tsem_t rspSem; // not used now
void* fp;
int64_t execId; // showId/queryId
SShowReqInfo showInfo; // todo this attribute will be removed after the query framework being completed.
SDataBuf requestMsg;
SReqResultInfo resInfo;
} SRequestSendRecvBody;
......@@ -121,6 +128,7 @@ typedef struct SRequestObj {
char *msgBuf;
void *pInfo; // sql parse info, generated by parser module
int32_t code;
uint64_t affectedRows;
SQueryExecMetric metric;
SRequestSendRecvBody body;
} SRequestObj;
......@@ -131,7 +139,7 @@ extern int32_t clientConnRefPool;
extern int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code);
SMsgSendInfo* buildSendMsgInfoImpl(SRequestObj*);
SMsgSendInfo* buildMsgInfoImpl(SRequestObj*);
int taos_init();
......
source/client/src/clientEnv.c
浏览文件 @ 60b56c66
......@@ -18,6 +18,7 @@
#include "clientInt.h"
#include "clientLog.h"
#include "query.h"
#include "scheduler.h"
#include "tmsg.h"
#include "tcache.h"
#include "tconfig.h"
......@@ -230,6 +231,8 @@ void taos_init_imp(void) {
SCatalogCfg cfg = {.maxDBCacheNum = 100, .maxTblCacheNum = 100};
catalogInit(&cfg);
SSchedulerCfg scfg = {.maxJobNum = 100};
schedulerInit(&scfg);
tscDebug("starting to initialize TAOS driver, local ep: %s", tsLocalEp);
taosSetCoreDump(true);
......
source/client/src/clientImpl.c
浏览文件 @ 60b56c66
......@@ -156,19 +156,12 @@ int32_t parseSql(SRequestObj* pRequest, SQueryNode** pQuery) {
};
cxt.ctx.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
// todo OPT performance
char buf[12] = {0};
sprintf(buf, "%"PRId64, pTscObj->pAppInfo->clusterId);
struct SCatalog* pCatalog = NULL;
int32_t code = catalogGetHandle(buf, &pCatalog);
int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.ctx.pCatalog);
if (code != TSDB_CODE_SUCCESS) {
tfree(cxt.ctx.db);
return code;
}
cxt.ctx.pCatalog = pCatalog;
code = qParseQuerySql(&cxt, pQuery);
tfree(cxt.ctx.db);
......@@ -181,10 +174,17 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
pRequest->body.requestMsg = (SDataBuf){.pData = pDcl->pMsg, .len = pDcl->msgLen};
STscObj* pTscObj = pRequest->pTscObj;
SMsgSendInfo* pSendMsg = buildSendMsgInfoImpl(pRequest);
SMsgSendInfo* pSendMsg = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
if (pDcl->msgType == TDMT_VND_CREATE_TABLE) {
if (pDcl->msgType == TDMT_VND_CREATE_TABLE || pDcl->msgType == TDMT_VND_SHOW_TABLES) {
if (pDcl->msgType == TDMT_VND_SHOW_TABLES) {
SShowReqInfo* pShowReqInfo = &pRequest->body.showInfo;
if (pShowReqInfo->pArray == NULL) {
pShowReqInfo->currentIndex = 0;
pShowReqInfo->pArray = pDcl->pExtension;
}
}
asyncSendMsgToServer(pTscObj->pTransporter, &pDcl->epSet, &transporterId, pSendMsg);
} else {
SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
......@@ -196,7 +196,15 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQueryNode* pQuery) {
return TSDB_CODE_SUCCESS;
}
int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQuery, SQueryDag** pDag) {
pRequest->type = pQuery->type;
return qCreateQueryDag(pQuery, pDag);
}
int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag, void** pJob) {
if (TSDB_SQL_INSERT == pRequest->type) {
return scheduleExecJob(pRequest->pTscObj->pTransporter, NULL/*todo appInfo.xxx*/, pDag, pJob, &pRequest->affectedRows);
}
return scheduleAsyncExecJob(pRequest->pTscObj->pTransporter, NULL/*todo appInfo.xxx*/, pDag, pJob);
}
......@@ -243,7 +251,7 @@ TAOS_RES *tmq_create_topic(TAOS* taos, const char* name, const char* sql, int sq
pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen };
SMsgSendInfo* body = buildSendMsgInfoImpl(pRequest);
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
SEpSet* pEpSet = &pTscObj->pAppInfo->mgmtEp.epSet;
int64_t transporterId = 0;
......@@ -283,7 +291,7 @@ TAOS_RES *taos_query_l(TAOS *taos, const char *sql, int sqlLen) {
if (qIsDdlQuery(pQuery)) {
CHECK_CODE_GOTO(execDdlQuery(pRequest, pQuery), _return);
} else {
CHECK_CODE_GOTO(qCreateQueryDag(pQuery, &pDag), _return);
CHECK_CODE_GOTO(getPlan(pRequest, pQuery, &pDag), _return);
CHECK_CODE_GOTO(scheduleQuery(pRequest, pDag, &pJob), _return);
}
......@@ -490,9 +498,38 @@ void* doFetchRow(SRequestObj* pRequest) {
SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
pRequest->type = TDMT_MND_SHOW_RETRIEVE;
if (pRequest->type == TDMT_MND_SHOW) {
pRequest->type = TDMT_MND_SHOW_RETRIEVE;
} else if (pRequest->type == TDMT_VND_SHOW_TABLES) {
pRequest->type = TDMT_VND_SHOW_TABLES_FETCH;
} else if (pRequest->type == TDMT_VND_SHOW_TABLES_FETCH) {
pRequest->type = TDMT_VND_SHOW_TABLES;
SShowReqInfo* pShowReqInfo = &pRequest->body.showInfo;
pShowReqInfo->currentIndex += 1;
if (pShowReqInfo->currentIndex >= taosArrayGetSize(pShowReqInfo->pArray)) {
return NULL;
}
SVgroupInfo* pVgroupInfo = taosArrayGet(pShowReqInfo->pArray, pShowReqInfo->currentIndex);
SVShowTablesReq* pShowReq = calloc(1, sizeof(SVShowTablesReq));
pShowReq->head.vgId = htonl(pVgroupInfo->vgId);
pRequest->body.requestMsg.len = sizeof(SVShowTablesReq);
pRequest->body.requestMsg.pData = pShowReq;
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
STscObj *pTscObj = pRequest->pTscObj;
asyncSendMsgToServer(pTscObj->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
tsem_wait(&pRequest->body.rspSem);
destroySendMsgInfo(body);
pRequest->type = TDMT_VND_SHOW_TABLES_FETCH;
}
SMsgSendInfo* body = buildSendMsgInfoImpl(pRequest);
SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
int64_t transporterId = 0;
STscObj *pTscObj = pRequest->pTscObj;
......
source/client/src/clientMsgHandler.c
浏览文件 @ 60b56c66
......@@ -18,23 +18,26 @@
#include "tname.h"
#include "clientInt.h"
#include "clientLog.h"
#include "trpc.h"
int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
static void setErrno(SRequestObj* pRequest, int32_t code) {
pRequest->code = code;
terrno = code;
}
int genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
pRequest->code = code;
setErrno(pRequest, code);
sem_post(&pRequest->body.rspSem);
return 0;
return code;
}
int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
terrno = code;
setErrno(pRequest, code);
sem_post(&pRequest->body.rspSem);
return code;
}
......@@ -74,52 +77,55 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
return 0;
}
static int32_t buildRetrieveMnodeMsg(SRequestObj *pRequest, SMsgSendInfo* pMsgSendInfo) {
pMsgSendInfo->msgType = TDMT_MND_SHOW_RETRIEVE;
pMsgSendInfo->msgInfo.len = sizeof(SRetrieveTableMsg);
pMsgSendInfo->requestObjRefId = pRequest->self;
pMsgSendInfo->param = pRequest;
pMsgSendInfo->fp = handleRequestRspFp[TMSG_INDEX(pMsgSendInfo->msgType)];
SRetrieveTableMsg *pRetrieveMsg = calloc(1, sizeof(SRetrieveTableMsg));
if (pRetrieveMsg == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
pRetrieveMsg->showId = htonl(pRequest->body.execId);
pMsgSendInfo->msgInfo.pData = pRetrieveMsg;
return TSDB_CODE_SUCCESS;
}
SMsgSendInfo* buildSendMsgInfoImpl(SRequestObj *pRequest) {
SMsgSendInfo* buildMsgInfoImpl(SRequestObj *pRequest) {
SMsgSendInfo* pMsgSendInfo = calloc(1, sizeof(SMsgSendInfo));
if (pRequest->type == TDMT_MND_SHOW_RETRIEVE) {
buildRetrieveMnodeMsg(pRequest, pMsgSendInfo);
pMsgSendInfo->requestObjRefId = pRequest->self;
pMsgSendInfo->requestId = pRequest->requestId;
pMsgSendInfo->param = pRequest;
pMsgSendInfo->msgType = pRequest->type;
if (pRequest->type == TDMT_MND_SHOW_RETRIEVE || pRequest->type == TDMT_VND_SHOW_TABLES_FETCH) {
if (pRequest->type == TDMT_MND_SHOW_RETRIEVE) {
SRetrieveTableMsg* pRetrieveMsg = calloc(1, sizeof(SRetrieveTableMsg));
if (pRetrieveMsg == NULL) {
return NULL;
}
pRetrieveMsg->showId = htobe64(pRequest->body.showInfo.execId);
pMsgSendInfo->msgInfo.pData = pRetrieveMsg;
pMsgSendInfo->msgInfo.len = sizeof(SRetrieveTableMsg);
} else {
SVShowTablesFetchReq* pFetchMsg = calloc(1, sizeof(SVShowTablesFetchReq));
if (pFetchMsg == NULL) {
return NULL;
}
pFetchMsg->id = htobe64(pRequest->body.showInfo.execId);
pFetchMsg->head.vgId = htonl(pRequest->body.showInfo.vgId);
pMsgSendInfo->msgInfo.pData = pFetchMsg;
pMsgSendInfo->msgInfo.len = sizeof(SVShowTablesFetchReq);
}
} else {
assert(pRequest != NULL);
pMsgSendInfo->requestObjRefId = pRequest->self;
pMsgSendInfo->msgInfo = pRequest->body.requestMsg;
pMsgSendInfo->msgType = pRequest->type;
pMsgSendInfo->requestId = pRequest->requestId;
pMsgSendInfo->param = pRequest;
pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)];
}
pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)];
return pMsgSendInfo;
}
int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
SShowRsp* pShow = (SShowRsp *)pMsg->pData;
pShow->showId = htonl(pShow->showId);
pShow->showId = htobe64(pShow->showId);
STableMetaMsg *pMetaMsg = &(pShow->tableMeta);
pMetaMsg->numOfColumns = htonl(pMetaMsg->numOfColumns);
......@@ -128,6 +134,7 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pMetaMsg->tuid = htobe64(pMetaMsg->tuid);
for (int i = 0; i < pMetaMsg->numOfColumns; ++i) {
pSchema->bytes = htonl(pSchema->bytes);
pSchema->colId = htonl(pSchema->colId);
pSchema++;
}
......@@ -148,34 +155,81 @@ int32_t processShowRsp(void* param, const SDataBuf* pMsg, int32_t code) {
pResInfo->pCol = calloc(pResInfo->numOfCols, POINTER_BYTES);
pResInfo->length = calloc(pResInfo->numOfCols, sizeof(int32_t));
pRequest->body.execId = pShow->showId;
pRequest->body.showInfo.execId = pShow->showId;
// todo
if (pRequest->type == TDMT_VND_SHOW_TABLES) {
SShowReqInfo* pShowInfo = &pRequest->body.showInfo;
int32_t index = pShowInfo->currentIndex;
SVgroupInfo* pInfo = taosArrayGet(pShowInfo->pArray, index);
pShowInfo->vgId = pInfo->vgId;
}
tsem_post(&pRequest->body.rspSem);
return 0;
}
int32_t processRetrieveMnodeRsp(void* param, const SDataBuf* pMsg, int32_t code) {
assert(pMsg->len >= sizeof(SRetrieveTableRsp));
SRequestObj *pRequest = param;
SReqResultInfo *pResInfo = &pRequest->body.resInfo;
tfree(pResInfo->pRspMsg);
SRequestObj* pRequest = param;
// tfree(pRequest->body.resInfo.pRspMsg);
// pRequest->body.resInfo.pRspMsg = pMsg->pData;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
assert(pMsg->len >= sizeof(SRetrieveTableRsp));
SRetrieveTableRsp *pRetrieve = (SRetrieveTableRsp *) pMsg->pData;
pRetrieve->numOfRows = htonl(pRetrieve->numOfRows);
pRetrieve->precision = htons(pRetrieve->precision);
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
tfree(pResInfo->pRspMsg);
pResInfo->pRspMsg = pMsg->pData;
pResInfo->numOfRows = pRetrieve->numOfRows;
pResInfo->pData = pRetrieve->data; // todo fix this in async model
pResInfo->pData = pRetrieve->data;
pResInfo->current = 0;
setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows);
tscDebug("0x%"PRIx64" numOfRows:%d, complete:%d, qId:0x%"PRIx64, pRequest->self, pRetrieve->numOfRows,
pRetrieve->completed, pRequest->body.execId);
pRetrieve->completed, pRequest->body.showInfo.execId);
tsem_post(&pRequest->body.rspSem);
return 0;
}
int32_t processRetrieveVndRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
SReqResultInfo* pResInfo = &pRequest->body.resInfo;
tfree(pResInfo->pRspMsg);
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
assert(pMsg->len >= sizeof(SRetrieveTableRsp));
pResInfo->pRspMsg = pMsg->pData;
SVShowTablesFetchRsp *pFetchRsp = (SVShowTablesFetchRsp *) pMsg->pData;
pFetchRsp->numOfRows = htonl(pFetchRsp->numOfRows);
pFetchRsp->precision = htons(pFetchRsp->precision);
pResInfo->pRspMsg = pMsg->pData;
pResInfo->numOfRows = pFetchRsp->numOfRows;
pResInfo->pData = pFetchRsp->data;
pResInfo->current = 0;
setResultDataPtr(pResInfo, pResInfo->fields, pResInfo->numOfCols, pResInfo->numOfRows);
tscDebug("0x%"PRIx64" numOfRows:%d, complete:%d, qId:0x%"PRIx64, pRequest->self, pFetchRsp->numOfRows,
pFetchRsp->completed, pRequest->body.showInfo.execId);
tsem_post(&pRequest->body.rspSem);
return 0;
......@@ -191,34 +245,48 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
SUseDbRsp* pUseDbRsp = (SUseDbRsp*)pMsg->pData;
SName name = {0};
tNameFromString(&name, pUseDbRsp->db, T_NAME_ACCT | T_NAME_DB);
SUseDbRsp* pUseDbRsp = (SUseDbRsp*) pMsg->pData;
SName name = {0};
tNameFromString(&name, pUseDbRsp->db, T_NAME_ACCT|T_NAME_DB);
char db[TSDB_DB_NAME_LEN] = {0};
tNameGetDbName(&name, db);
setConnectionDB(pRequest->pTscObj, db);
tsem_post(&pRequest->body.rspSem);
return 0;
}
int32_t processCreateTableRsp(void* param, const SDataBuf* pMsg, int32_t code) {
assert(pMsg != NULL);
assert(pMsg != NULL && param != NULL);
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
tsem_post(&pRequest->body.rspSem);
return code;
}
int32_t processDropDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
// todo: Remove cache in catalog cache.
SRequestObj* pRequest = param;
if (code != TSDB_CODE_SUCCESS) {
setErrno(pRequest, code);
tsem_post(&pRequest->body.rspSem);
return code;
}
tsem_post(&pRequest->body.rspSem);
return code;
}
void initMsgHandleFp() {
......@@ -304,4 +372,7 @@ void initMsgHandleFp() {
handleRequestRspFp[TMSG_INDEX(TDMT_MND_USE_DB)] = processUseDbRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_MND_CREATE_STB)] = processCreateTableRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = processDropDbRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = processShowRsp;
handleRequestRspFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = processRetrieveVndRsp;
}
\ No newline at end of file
source/client/test/clientTests.cpp
浏览文件 @ 60b56c66
此差异已折叠。
source/common/src/tmsg.c
浏览文件 @ 60b56c66
......@@ -25,4 +25,230 @@
#undef TD_MSG_INFO_
#define TD_MSG_DICT_
#undef TD_MSG_SEG_CODE_
#include "tmsgdef.h"
\ No newline at end of file
#include "tmsgdef.h"
static int tmsgStartEncode(SMsgEncoder *pME);
static void tmsgEndEncode(SMsgEncoder *pME);
static int tmsgStartDecode(SMsgDecoder *pMD);
static void tmsgEndDecode(SMsgDecoder *pMD);
/* ------------------------ ENCODE/DECODE FUNCTIONS ------------------------ */
void tmsgInitMsgEncoder(SMsgEncoder *pME, td_endian_t endian, uint8_t *data, int64_t size) {
tInitEncoder(&(pME->coder), endian, data, size);
TD_SLIST_INIT(&(pME->eStack));
}
void tmsgClearMsgEncoder(SMsgEncoder *pME) {
struct SMEListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pME->eStack));
if (TD_IS_NULL(pNode)) break;
TD_SLIST_POP(&(pME->eStack));
free(pNode);
}
}
void tmsgInitMsgDecoder(SMsgDecoder *pMD, td_endian_t endian, uint8_t *data, int64_t size) {
tInitDecoder(&pMD->coder, endian, data, size);
TD_SLIST_INIT(&(pMD->dStack));
TD_SLIST_INIT(&(pMD->freeList));
}
void tmsgClearMsgDecoder(SMsgDecoder *pMD) {
{
struct SMDFreeListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pMD->freeList));
if (TD_IS_NULL(pNode)) break;
TD_SLIST_POP(&(pMD->freeList));
free(pNode);
}
}
{
struct SMDListNode *pNode;
for (;;) {
pNode = TD_SLIST_HEAD(&(pMD->dStack));
if (TD_IS_NULL(pNode)) break;
TD_SLIST_POP(&(pMD->dStack));
free(pNode);
}
}
}
/* ------------------------ MESSAGE ENCODE/DECODE ------------------------ */
int tmsgSVCreateTbReqEncode(SMsgEncoder *pCoder, SVCreateTbReq *pReq) {
tmsgStartEncode(pCoder);
// TODO
tmsgEndEncode(pCoder);
return 0;
}
int tmsgSVCreateTbReqDecode(SMsgDecoder *pCoder, SVCreateTbReq *pReq) {
tmsgStartDecode(pCoder);
// TODO: decode
// Decode is not end
if (pCoder->coder.pos != pCoder->coder.size) {
// Continue decode
}
tmsgEndDecode(pCoder);
return 0;
}
int tSerializeSVCreateTbReq(void **buf, const SVCreateTbReq *pReq) {
int tlen = 0;
tlen += taosEncodeFixedU64(buf, pReq->ver);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedU32(buf, pReq->ttl);
tlen += taosEncodeFixedU32(buf, pReq->keep);
tlen += taosEncodeFixedU8(buf, pReq->type);
switch (pReq->type) {
case TD_SUPER_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->stbCfg.suid);
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nCols);
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pSchema[i].name);
}
tlen += taosEncodeFixedU32(buf, pReq->stbCfg.nTagCols);
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->stbCfg.pTagSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->stbCfg.pTagSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
tlen += taosEncodeFixedU64(buf, pReq->ctbCfg.suid);
tlen += tdEncodeKVRow(buf, pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
tlen += taosEncodeFixedU32(buf, pReq->ntbCfg.nCols);
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
tlen += taosEncodeFixedI8(buf, pReq->ntbCfg.pSchema[i].type);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].colId);
tlen += taosEncodeFixedI32(buf, pReq->ntbCfg.pSchema[i].bytes);
tlen += taosEncodeString(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return tlen;
}
void *tDeserializeSVCreateTbReq(void *buf, SVCreateTbReq *pReq) {
buf = taosDecodeFixedU64(buf, &(pReq->ver));
buf = taosDecodeString(buf, &(pReq->name));
buf = taosDecodeFixedU32(buf, &(pReq->ttl));
buf = taosDecodeFixedU32(buf, &(pReq->keep));
buf = taosDecodeFixedU8(buf, &(pReq->type));
switch (pReq->type) {
case TD_SUPER_TABLE:
buf = taosDecodeFixedU64(buf, &(pReq->stbCfg.suid));
buf = taosDecodeFixedU32(buf, &(pReq->stbCfg.nCols));
pReq->stbCfg.pSchema = (SSchema *)malloc(pReq->stbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pSchema[i].type));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].colId));
buf = taosDecodeFixedI32(buf, &(pReq->stbCfg.pSchema[i].bytes));
buf = taosDecodeStringTo(buf, pReq->stbCfg.pSchema[i].name);
}
buf = taosDecodeFixedU32(buf, &pReq->stbCfg.nTagCols);
pReq->stbCfg.pTagSchema = (SSchema *)malloc(pReq->stbCfg.nTagCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->stbCfg.nTagCols; i++) {
buf = taosDecodeFixedI8(buf, &(pReq->stbCfg.pTagSchema[i].type));
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->stbCfg.pTagSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->stbCfg.pTagSchema[i].name);
}
break;
case TD_CHILD_TABLE:
buf = taosDecodeFixedU64(buf, &pReq->ctbCfg.suid);
buf = tdDecodeKVRow(buf, &pReq->ctbCfg.pTag);
break;
case TD_NORMAL_TABLE:
buf = taosDecodeFixedU32(buf, &pReq->ntbCfg.nCols);
pReq->ntbCfg.pSchema = (SSchema *)malloc(pReq->ntbCfg.nCols * sizeof(SSchema));
for (uint32_t i = 0; i < pReq->ntbCfg.nCols; i++) {
buf = taosDecodeFixedI8(buf, &pReq->ntbCfg.pSchema[i].type);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].colId);
buf = taosDecodeFixedI32(buf, &pReq->ntbCfg.pSchema[i].bytes);
buf = taosDecodeStringTo(buf, pReq->ntbCfg.pSchema[i].name);
}
break;
default:
ASSERT(0);
}
return buf;
}
/* ------------------------ STATIC METHODS ------------------------ */
static int tmsgStartEncode(SMsgEncoder *pME) {
struct SMEListNode *pNode = (struct SMEListNode *)malloc(sizeof(*pNode));
if (TD_IS_NULL(pNode)) return -1;
pNode->coder = pME->coder;
TD_SLIST_PUSH(&(pME->eStack), pNode);
TD_CODER_MOVE_POS(&(pME->coder), sizeof(int32_t));
return 0;
}
static void tmsgEndEncode(SMsgEncoder *pME) {
int32_t size;
struct SMEListNode *pNode;
pNode = TD_SLIST_HEAD(&(pME->eStack));
ASSERT(pNode);
TD_SLIST_POP(&(pME->eStack));
size = pME->coder.pos - pNode->coder.pos;
tEncodeI32(&(pNode->coder), size);
free(pNode);
}
static int tmsgStartDecode(SMsgDecoder *pMD) {
struct SMDListNode *pNode;
int32_t size;
pNode = (struct SMDListNode *)malloc(sizeof(*pNode));
if (pNode == NULL) return -1;
tDecodeI32(&(pMD->coder), &size);
pNode->coder = pMD->coder;
TD_SLIST_PUSH(&(pMD->dStack), pNode);
pMD->coder.pos = 0;
pMD->coder.size = size - sizeof(int32_t);
pMD->coder.data = TD_CODER_CURRENT(&(pNode->coder));
return 0;
}
static void tmsgEndDecode(SMsgDecoder *pMD) {
ASSERT(pMD->coder.pos == pMD->coder.size);
struct SMDListNode *pNode;
pNode = TD_SLIST_HEAD(&(pMD->dStack));
ASSERT(pNode);
TD_SLIST_POP(&(pMD->dStack));
pNode->coder.pos += pMD->coder.size;
pMD->coder = pNode->coder;
free(pNode);
}
\ No newline at end of file
source/common/src/tname.c
浏览文件 @ 60b56c66
......@@ -259,3 +259,13 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
return 0;
}
SSchema createSchema(uint8_t type, int32_t bytes, int32_t colId, const char* name) {
SSchema s = {0};
s.type = type;
s.bytes = bytes;
s.colId = colId;
tstrncpy(s.name, name, tListLen(s.name));
return s;
}
\ No newline at end of file
source/common/test/CMakeLists.txt
浏览文件 @ 60b56c66
......@@ -18,11 +18,11 @@ TARGET_INCLUDE_DIRECTORIES(
)
# tmsg test
add_executable(tmsgTest "")
target_sources(tmsgTest
PRIVATE
"tmsgTest.cpp"
"../src/tmsg.c"
)
target_include_directories(tmsgTest PUBLIC "${CMAKE_SOURCE_DIR}/include/common/")
target_link_libraries(tmsgTest PUBLIC os util gtest gtest_main)
\ No newline at end of file
# add_executable(tmsgTest "")
# target_sources(tmsgTest
# PRIVATE
# "tmsgTest.cpp"
# "../src/tmsg.c"
# )
# target_include_directories(tmsgTest PUBLIC "${CMAKE_SOURCE_DIR}/include/common/")
# target_link_libraries(tmsgTest PUBLIC os util gtest gtest_main)
\ No newline at end of file
source/dnode/mgmt/impl/src/dndTransport.c
浏览文件 @ 60b56c66
......@@ -141,6 +141,8 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CREATE_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_DROP_TABLE)] = dndProcessVnodeWriteMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = dndProcessVnodeFetchMsg;
pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = dndProcessVnodeFetchMsg;
}
static void dndProcessResponse(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
......
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ 60b56c66
......@@ -285,7 +285,7 @@ typedef struct {
} SFuncObj;
typedef struct {
int32_t id;
int64_t id;
int8_t type;
int8_t replica;
int16_t numOfColumns;
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ 60b56c66
......@@ -41,7 +41,7 @@ typedef struct {
} SMnodeStep;
typedef struct {
int32_t showId;
int64_t showId;
ShowMetaFp metaFps[TSDB_MGMT_TABLE_MAX];
ShowRetrieveFp retrieveFps[TSDB_MGMT_TABLE_MAX];
ShowFreeIterFp freeIterFps[TSDB_MGMT_TABLE_MAX];
......
source/dnode/mnode/impl/src/mndShow.c
浏览文件 @ 60b56c66
......@@ -20,7 +20,7 @@
static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowMsg *pMsg);
static void mndFreeShowObj(SShowObj *pShow);
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int32_t showId);
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int64_t showId);
static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove);
static int32_t mndProcessShowMsg(SMnodeMsg *pMnodeMsg);
static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMsg);
......@@ -52,8 +52,8 @@ void mndCleanupShow(SMnode *pMnode) {
static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowMsg *pMsg) {
SShowMgmt *pMgmt = &pMnode->showMgmt;
int32_t showId = atomic_add_fetch_32(&pMgmt->showId, 1);
if (showId == 0) atomic_add_fetch_32(&pMgmt->showId, 1);
int64_t showId = atomic_add_fetch_64(&pMgmt->showId, 1);
if (showId == 0) atomic_add_fetch_64(&pMgmt->showId, 1);
int32_t size = sizeof(SShowObj) + pMsg->payloadLen;
SShowObj showObj = {0};
......@@ -65,14 +65,14 @@ static SShowObj *mndCreateShowObj(SMnode *pMnode, SShowMsg *pMsg) {
memcpy(showObj.payload, pMsg->payload, pMsg->payloadLen);
int32_t keepTime = pMnode->cfg.shellActivityTimer * 6 * 1000;
SShowObj *pShow = taosCachePut(pMgmt->cache, &showId, sizeof(int32_t), &showObj, size, keepTime);
SShowObj *pShow = taosCachePut(pMgmt->cache, &showId, sizeof(int64_t), &showObj, size, keepTime);
if (pShow == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("show:%d, failed to put into cache since %s", showId, terrstr());
mError("show:0x%"PRIx64", failed to put into cache since %s", showId, terrstr());
return NULL;
}
mTrace("show:%d, is created, data:%p", showId, pShow);
mTrace("show:0x%"PRIx64", is created, data:%p", showId, pShow);
return pShow;
}
......@@ -87,25 +87,25 @@ static void mndFreeShowObj(SShowObj *pShow) {
}
}
mTrace("show:%d, is destroyed, data:%p", pShow->id, pShow);
mTrace("show:0x%d, is destroyed, data:%p", pShow->id, pShow);
}
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int32_t showId) {
static SShowObj *mndAcquireShowObj(SMnode *pMnode, int64_t showId) {
SShowMgmt *pMgmt = &pMnode->showMgmt;
SShowObj *pShow = taosCacheAcquireByKey(pMgmt->cache, &showId, sizeof(int32_t));
SShowObj *pShow = taosCacheAcquireByKey(pMgmt->cache, &showId, sizeof(showId));
if (pShow == NULL) {
mError("show:%d, already destroyed", showId);
mError("show:0x%"PRIx64", already destroyed", showId);
return NULL;
}
mTrace("show:%d, acquired from cache, data:%p", pShow->id, pShow);
mTrace("show:0x%"PRIx64", acquired from cache, data:%p", pShow->id, pShow);
return pShow;
}
static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove) {
if (pShow == NULL) return;
mTrace("show:%d, released from cache, data:%p force:%d", pShow->id, pShow, forceRemove);
mTrace("show:0x%"PRIx64", released from cache, data:%p force:%d", pShow->id, pShow, forceRemove);
// A bug in tcache.c
forceRemove = 0;
......@@ -146,18 +146,18 @@ static int32_t mndProcessShowMsg(SMnodeMsg *pMnodeMsg) {
if (pRsp == NULL) {
mndReleaseShowObj(pShow, true);
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("show:%d, failed to process show-meta msg:%s since malloc rsp error", pShow->id, mndShowStr(type));
mError("show:0x%"PRIx64", failed to process show-meta msg:%s since malloc rsp error", pShow->id, mndShowStr(type));
return -1;
}
int32_t code = (*metaFp)(pMnodeMsg, pShow, &pRsp->tableMeta);
mDebug("show:%d, get meta finished, numOfRows:%d cols:%d type:%s result:%s", pShow->id, pShow->numOfRows,
mDebug("show:0x%"PRIx64", get meta finished, numOfRows:%d cols:%d type:%s result:%s", pShow->id, pShow->numOfRows,
pShow->numOfColumns, mndShowStr(type), tstrerror(code));
if (code == TSDB_CODE_SUCCESS) {
pMnodeMsg->contLen = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
pMnodeMsg->pCont = pRsp;
pRsp->showId = htonl(pShow->id);
pMnodeMsg->pCont = pRsp;
pRsp->showId = htobe64(pShow->id);
mndReleaseShowObj(pShow, false);
return TSDB_CODE_SUCCESS;
} else {
......@@ -175,7 +175,7 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
int32_t rowsRead = 0;
SRetrieveTableMsg *pRetrieve = pMnodeMsg->rpcMsg.pCont;
int32_t showId = htonl(pRetrieve->showId);
int64_t showId = htobe64(pRetrieve->showId);
SShowObj *pShow = mndAcquireShowObj(pMnode, showId);
if (pShow == NULL) {
......@@ -188,15 +188,15 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
if (retrieveFp == NULL) {
mndReleaseShowObj(pShow, false);
terrno = TSDB_CODE_MSG_NOT_PROCESSED;
mError("show:%d, failed to retrieve data since %s", pShow->id, terrstr());
mError("show:0x%"PRIx64", failed to retrieve data since %s", pShow->id, terrstr());
return -1;
}
mDebug("show:%d, start retrieve data, numOfReads:%d numOfRows:%d type:%s", pShow->id, pShow->numOfReads,
mDebug("show:0x%"PRIx64", start retrieve data, numOfReads:%d numOfRows:%d type:%s", pShow->id, pShow->numOfReads,
pShow->numOfRows, mndShowStr(pShow->type));
if (mndCheckRetrieveFinished(pShow)) {
mDebug("show:%d, read finished, numOfReads:%d numOfRows:%d", pShow->id, pShow->numOfReads, pShow->numOfRows);
mDebug("show:0x%"PRIx64", read finished, numOfReads:%d numOfRows:%d", pShow->id, pShow->numOfReads, pShow->numOfRows);
pShow->numOfReads = pShow->numOfRows;
}
......@@ -219,7 +219,7 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
if (pRsp == NULL) {
mndReleaseShowObj(pShow, false);
terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("show:%d, failed to retrieve data since %s", pShow->id, terrstr());
mError("show:0x%"PRIx64", failed to retrieve data since %s", pShow->id, terrstr());
return -1;
}
......@@ -228,7 +228,7 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
rowsRead = (*retrieveFp)(pMnodeMsg, pShow, pRsp->data, rowsToRead);
}
mDebug("show:%d, stop retrieve data, rowsRead:%d rowsToRead:%d", pShow->id, rowsRead, rowsToRead);
mDebug("show:0x%"PRIx64", stop retrieve data, rowsRead:%d rowsToRead:%d", pShow->id, rowsRead, rowsToRead);
pRsp->numOfRows = htonl(rowsRead);
pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision
......@@ -238,10 +238,10 @@ static int32_t mndProcessRetrieveMsg(SMnodeMsg *pMnodeMsg) {
if (rowsRead == 0 || rowsToRead == 0 || (rowsRead == rowsToRead && pShow->numOfRows == pShow->numOfReads)) {
pRsp->completed = 1;
mDebug("show:%d, retrieve completed", pShow->id);
mDebug("show:0x%"PRIx64", retrieve completed", pShow->id);
mndReleaseShowObj(pShow, true);
} else {
mDebug("show:%d, retrieve not completed yet", pShow->id);
mDebug("show:0x%"PRIx64", retrieve not completed yet", pShow->id);
mndReleaseShowObj(pShow, false);
}
......
source/dnode/vnode/impl/inc/vnodeDef.h
浏览文件 @ 60b56c66
......@@ -31,10 +31,10 @@
#include "vnodeCommit.h"
#include "vnodeFS.h"
#include "vnodeMemAllocator.h"
#include "vnodeQuery.h"
#include "vnodeRequest.h"
#include "vnodeStateMgr.h"
#include "vnodeSync.h"
#include "vnodeQuery.h"
#ifdef __cplusplus
extern "C" {
......@@ -62,6 +62,7 @@ typedef struct SVnodeMgr {
extern SVnodeMgr vnodeMgr;
struct SVnode {
int32_t vgId;
char* path;
SVnodeCfg config;
SVState state;
......
source/dnode/vnode/impl/src/vnodeQuery.c
浏览文件 @ 60b56c66
......@@ -16,6 +16,8 @@
#include "vnodeQuery.h"
#include "vnodeDef.h"
static int32_t vnodeGetTableList(SVnode *pVnode, SRpcMsg *pMsg);
int vnodeQueryOpen(SVnode *pVnode) { return qWorkerInit(NULL, &pVnode->pQuery); }
int vnodeProcessQueryReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
......@@ -36,32 +38,132 @@ int vnodeProcessFetchReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
return qWorkerProcessCancelMsg(pVnode, pVnode->pQuery, pMsg);
case TDMT_VND_DROP_TASK:
return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg);
case TDMT_VND_SHOW_TABLES:
return qWorkerProcessShowMsg(pVnode, pVnode->pQuery, pMsg);
case TDMT_VND_SHOW_TABLES_FETCH:
return vnodeGetTableList(pVnode, pMsg);
// return qWorkerProcessShowFetchMsg(pVnode->pMeta, pVnode->pQuery, pMsg);
default:
vError("unknown msg type:%d in fetch queue", pMsg->msgType);
return TSDB_CODE_VND_APP_ERROR;
break;
}
return 0;
}
static int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
STableInfoMsg *pReq = (STableInfoMsg *)(pMsg->pCont);
STableMetaMsg *pRspMsg;
int ret;
STableInfoMsg * pReq = (STableInfoMsg *)(pMsg->pCont);
STbCfg * pTbCfg = NULL;
STbCfg * pStbCfg = NULL;
tb_uid_t uid;
int32_t nCols;
int32_t nTagCols;
SSchemaWrapper *pSW;
STableMetaMsg * pTbMetaMsg;
SSchema * pTagSchema;
if (metaGetTableInfo(pVnode->pMeta, pReq->tableFname, &pRspMsg) < 0) {
pTbCfg = metaGetTbInfoByName(pVnode->pMeta, pReq->tableFname, &uid);
if (pTbCfg == NULL) {
return -1;
}
*pRsp = malloc(sizeof(SRpcMsg));
if (TD_IS_NULL(*pRsp)) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
free(pMsg);
if (pTbCfg->type == META_CHILD_TABLE) {
pStbCfg = metaGetTbInfoByUid(pVnode->pMeta, pTbCfg->ctbCfg.suid);
if (pStbCfg == NULL) {
return -1;
}
pSW = metaGetTableSchema(pVnode->pMeta, pTbCfg->ctbCfg.suid, 0, true);
} else {
pSW = metaGetTableSchema(pVnode->pMeta, uid, 0, true);
}
nCols = pSW->nCols;
if (pTbCfg->type == META_SUPER_TABLE) {
nTagCols = pTbCfg->stbCfg.nTagCols;
pTagSchema = pTbCfg->stbCfg.pTagSchema;
} else if (pTbCfg->type == META_SUPER_TABLE) {
nTagCols = pStbCfg->stbCfg.nTagCols;
pTagSchema = pStbCfg->stbCfg.pTagSchema;
} else {
nTagCols = 0;
pTagSchema = NULL;
}
pTbMetaMsg = (STableMetaMsg *)calloc(1, sizeof(STableMetaMsg) + sizeof(SSchema) * (nCols + nTagCols));
if (pTbMetaMsg == NULL) {
return -1;
}
// TODO
(*pRsp)->pCont = pRspMsg;
strcpy(pTbMetaMsg->tbFname, pTbCfg->name);
if (pTbCfg->type == META_CHILD_TABLE) {
strcpy(pTbMetaMsg->stbFname, pStbCfg->name);
pTbMetaMsg->suid = htobe64(pTbCfg->ctbCfg.suid);
}
pTbMetaMsg->numOfTags = htonl(nTagCols);
pTbMetaMsg->numOfColumns = htonl(nCols);
pTbMetaMsg->tableType = pTbCfg->type;
pTbMetaMsg->tuid = htobe64(uid);
pTbMetaMsg->vgId = htonl(pVnode->vgId);
memcpy(pTbMetaMsg->pSchema, pSW->pSchema, sizeof(SSchema) * pSW->nCols);
if (nTagCols) {
memcpy(POINTER_SHIFT(pTbMetaMsg->pSchema, sizeof(SSchema) * pSW->nCols), pTagSchema, sizeof(SSchema) * nTagCols);
}
for (int i = 0; i < nCols + nTagCols; i++) {
SSchema *pSch = pTbMetaMsg->pSchema + i;
pSch->colId = htonl(pSch->colId);
pSch->bytes = htonl(pSch->bytes);
}
return 0;
}
/**
* @param pVnode
* @param pMsg
* @param pRsp
*/
static int32_t vnodeGetTableList(SVnode *pVnode, SRpcMsg *pMsg) {
SMTbCursor* pCur = metaOpenTbCursor(pVnode->pMeta);
SArray* pArray = taosArrayInit(10, POINTER_BYTES);
char* name = NULL;
int32_t totalLen = 0;
while ((name = metaTbCursorNext(pCur)) != NULL) {
taosArrayPush(pArray, &name);
totalLen += strlen(name);
}
metaCloseTbCursor(pCur);
int32_t rowLen = (TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE) + 8 + 4 + (TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE) + 8 + 4;
int32_t numOfTables = (int32_t) taosArrayGetSize(pArray);
int32_t payloadLen = rowLen * numOfTables;
// SVShowTablesFetchReq *pFetchReq = pMsg->pCont;
SVShowTablesFetchRsp *pFetchRsp = (SVShowTablesFetchRsp *)rpcMallocCont(sizeof(SVShowTablesFetchRsp) + payloadLen);
memset(pFetchRsp, 0, sizeof(struct SVShowTablesFetchRsp) + payloadLen);
char* p = pFetchRsp->data;
for(int32_t i = 0; i < numOfTables; ++i) {
char* n = taosArrayGetP(pArray, i);
STR_TO_VARSTR(p, n);
p += rowLen;
}
pFetchRsp->numOfRows = htonl(numOfTables);
pFetchRsp->precision = 0;
SRpcMsg rpcMsg = {
.handle = pMsg->handle,
.ahandle = pMsg->ahandle,
.pCont = pFetchRsp,
.contLen = sizeof(SVShowTablesFetchRsp) + payloadLen,
.code = 0,
};
rpcSendResponse(&rpcMsg);
return 0;
}
\ No newline at end of file
source/dnode/vnode/meta/src/metaBDBImpl.c
浏览文件 @ 60b56c66
......@@ -431,79 +431,154 @@ static void metaClearTbCfg(STbCfg *pTbCfg) {
}
/* ------------------------ FOR QUERY ------------------------ */
int metaGetTableInfo(SMeta *pMeta, char *tbname, STableMetaMsg **ppMsg) {
DBT key = {0};
DBT value = {0};
SMetaDB * pMetaDB = pMeta->pDB;
int ret;
STbCfg tbCfg;
SSchemaKey schemaKey;
DBT key1 = {0};
DBT value1 = {0};
uint32_t ncols;
void * pBuf;
int tlen;
STableMetaMsg *pMsg;
STbCfg *metaGetTbInfoByUid(SMeta *pMeta, tb_uid_t uid) {
STbCfg * pTbCfg = NULL;
SMetaDB *pDB = pMeta->pDB;
DBT key = {0};
DBT value = {0};
int ret;
// Set key/value
key.data = &uid;
key.size = sizeof(uid);
// Query
ret = pDB->pTbDB->get(pDB->pTbDB, NULL, &key, &value, 0);
if (ret != 0) {
return NULL;
}
// Decode
pTbCfg = (STbCfg *)malloc(sizeof(*pTbCfg));
if (pTbCfg == NULL) {
return NULL;
}
metaDecodeTbInfo(value.data, pTbCfg);
return pTbCfg;
}
STbCfg *metaGetTbInfoByName(SMeta *pMeta, char *tbname, tb_uid_t *uid) {
STbCfg * pTbCfg = NULL;
SMetaDB *pDB = pMeta->pDB;
DBT key = {0};
DBT pkey = {0};
DBT pvalue = {0};
int ret;
// Set key/value
key.data = tbname;
key.size = strlen(tbname) + 1;
key.size = strlen(tbname);
ret = pMetaDB->pNameIdx->get(pMetaDB->pNameIdx, NULL, &key, &value, 0);
// Query
ret = pDB->pNameIdx->pget(pDB->pNameIdx, NULL, &key, &pkey, &pvalue, 0);
if (ret != 0) {
// TODO
return -1;
return NULL;
}
metaDecodeTbInfo(value.data, &tbCfg);
switch (tbCfg.type) {
case META_SUPER_TABLE:
schemaKey.uid = tbCfg.stbCfg.suid;
schemaKey.sver = 0;
key1.data = &schemaKey;
key1.size = sizeof(schemaKey);
ret = pMetaDB->pSchemaDB->get(pMetaDB->pSchemaDB, &key1, &value1, NULL, 0);
if (ret != 0) {
// TODO
return -1;
}
pBuf = value1.data;
pBuf = taosDecodeFixedU32(pBuf, &ncols);
tlen = sizeof(STableMetaMsg) + (tbCfg.stbCfg.nTagCols + ncols) * sizeof(SSchema);
pMsg = calloc(1, tlen);
if (pMsg == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
strcpy(pMsg->tbFname, tbCfg.name);
pMsg->numOfTags = tbCfg.stbCfg.nTagCols;
pMsg->numOfColumns = ncols;
pMsg->tableType = tbCfg.type;
pMsg->sversion = 0;
pMsg->tversion = 0;
pMsg->suid = tbCfg.stbCfg.suid;
pMsg->tuid = tbCfg.stbCfg.suid;
for (size_t i = 0; i < tbCfg.stbCfg.nTagCols; i++) {
}
break;
case META_CHILD_TABLE:
ASSERT(0);
break;
case META_NORMAL_TABLE:
ASSERT(0);
break;
default:
ASSERT(0);
break;
}
*ppMsg = pMsg;
// Decode
*uid = *(tb_uid_t *)(pkey.data);
pTbCfg = (STbCfg *)malloc(sizeof(*pTbCfg));
if (pTbCfg == NULL) {
return NULL;
}
return 0;
metaDecodeTbInfo(pvalue.data, pTbCfg);
return pTbCfg;
}
SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
uint32_t nCols;
SSchemaWrapper *pSW = NULL;
SMetaDB * pDB = pMeta->pDB;
int ret;
void * pBuf;
SSchema * pSchema;
SSchemaKey schemaKey = {uid, sver};
DBT key = {0};
DBT value = {0};
// Set key/value properties
key.data = &schemaKey;
key.size = sizeof(schemaKey);
// Query
ret = pDB->pSchemaDB->get(pDB->pSchemaDB, NULL, &key, &value, 0);
if (ret != 0) {
return NULL;
}
// Decode the schema
pBuf = value.data;
taosDecodeFixedI32(&pBuf, &nCols);
if (isinline) {
pSW = (SSchemaWrapper *)malloc(sizeof(*pSW) + sizeof(SSchema) * nCols);
if (pSW == NULL) {
return NULL;
}
pSW->pSchema = POINTER_SHIFT(pSW, sizeof(*pSW));
} else {
pSW = (SSchemaWrapper *)malloc(sizeof(*pSW));
if (pSW == NULL) {
return NULL;
}
pSW->pSchema = (SSchema *)malloc(sizeof(SSchema) * nCols);
if (pSW->pSchema == NULL) {
free(pSW);
return NULL;
}
}
for (int i = 0; i < nCols; i++) {
pSchema = pSW->pSchema + i;
taosDecodeFixedI8(&pBuf, &(pSchema->type));
taosDecodeFixedI32(&pBuf, &(pSchema->colId));
taosDecodeFixedI32(&pBuf, &(pSchema->bytes));
taosDecodeStringTo(&pBuf, pSchema->name);
}
return pSW;
}
struct SMTbCursor {
DBC *pCur;
};
SMTbCursor *metaOpenTbCursor(SMeta *pMeta) {
SMTbCursor *pTbCur = NULL;
SMetaDB * pDB = pMeta->pDB;
pTbCur = (SMTbCursor *)calloc(1, sizeof(*pTbCur));
if (pTbCur == NULL) {
return NULL;
}
pDB->pTbDB->cursor(pDB->pTbDB, NULL, &(pTbCur->pCur), 0);
return pTbCur;
}
void metaCloseTbCursor(SMTbCursor *pTbCur) {
if (pTbCur) {
if (pTbCur->pCur) {
pTbCur->pCur->close(pTbCur->pCur);
}
free(pTbCur);
}
}
char *metaTbCursorNext(SMTbCursor *pTbCur) {
DBT key = {0};
DBT value = {0};
STbCfg tbCfg;
if (pTbCur->pCur->get(pTbCur->pCur, &key, &value, DB_NEXT) == 0) {
metaDecodeTbInfo(&(value.data), &tbCfg);
return tbCfg.name;
} else {
return NULL;
}
}
\ No newline at end of file
source/libs/catalog/src/catalog.c
浏览文件 @ 60b56c66
......@@ -465,7 +465,7 @@ int32_t catalogInit(SCatalogCfg *cfg) {
ctgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TABLEMETA_NUMBER;
}
ctgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
ctgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
if (NULL == ctgMgmt.pCluster) {
CTG_ERR_LRET(TSDB_CODE_CTG_INTERNAL_ERROR, "init %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
}
......@@ -473,8 +473,8 @@ int32_t catalogInit(SCatalogCfg *cfg) {
return TSDB_CODE_SUCCESS;
}
int32_t catalogGetHandle(const char* clusterId , struct SCatalog** catalogHandle) {
if (NULL == clusterId || NULL == catalogHandle) {
int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle) {
if (NULL == catalogHandle) {
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
......@@ -483,8 +483,7 @@ int32_t catalogGetHandle(const char* clusterId , struct SCatalog** catalogHandle
CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
}
size_t clen = strlen(clusterId);
SCatalog **ctg = (SCatalog **)taosHashGet(ctgMgmt.pCluster, clusterId, clen);
SCatalog **ctg = (SCatalog **)taosHashGet(ctgMgmt.pCluster, (char*)&clusterId, sizeof(clusterId));
if (ctg && (*ctg)) {
*catalogHandle = *ctg;
......@@ -497,8 +496,8 @@ int32_t catalogGetHandle(const char* clusterId , struct SCatalog** catalogHandle
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
}
if (taosHashPut(ctgMgmt.pCluster, clusterId, clen, &clusterCtg, POINTER_BYTES)) {
ctgError("put cluster %s cache to hash failed", clusterId);
if (taosHashPut(ctgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES)) {
ctgError("put cluster %"PRIx64" cache to hash failed", clusterId);
tfree(clusterCtg);
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
......
source/libs/catalog/test/catalogTests.cpp
浏览文件 @ 60b56c66
......@@ -47,7 +47,7 @@ int32_t ctgTestTagNum = 1;
int32_t ctgTestSVersion = 1;
int32_t ctgTestTVersion = 1;
char *ctgTestClusterId = "cluster1";
uint64_t ctgTestClusterId = 0x1;
char *ctgTestDbname = "1.db1";
char *ctgTestTablename = "table1";
char *ctgTestCTablename = "ctable1";
......
source/libs/parser/inc/parserInt.h
浏览文件 @ 60b56c66
......@@ -102,7 +102,7 @@ int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SCatalogReq* p
* Destroy the meta data request structure.
* @param pMetaInfo
*/
void qParserClearupMetaRequestInfo(SCatalogReq* pMetaInfo);
void qParserCleanupMetaRequestInfo(SCatalogReq* pMetaInfo);
#ifdef __cplusplus
}
......
source/libs/parser/src/astToMsg.c
浏览文件 @ 60b56c66
......@@ -26,15 +26,13 @@ SCreateUserMsg* buildUserManipulationMsg(SSqlInfo* pInfo, int32_t* outputLen, in
}
SCreateAcctMsg* buildAcctManipulationMsg(SSqlInfo* pInfo, int32_t* outputLen, int64_t id, char* msgBuf, int32_t msgLen) {
SCreateAcctMsg* pMsg = (SCreateAcctMsg*)calloc(1, sizeof(SCreateAcctMsg));
if (pMsg == NULL) {
// tscError("0x%" PRIx64 " failed to malloc for query msg", id);
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
SCreateAcctMsg *pCreateMsg = (SCreateAcctMsg *) calloc(1, sizeof(SCreateAcctMsg));
if (pCreateMsg == NULL) {
qError("0x%" PRIx64 " failed to malloc for query msg", id);
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return NULL;
}
SCreateAcctMsg *pCreateMsg = (SCreateAcctMsg *) calloc(1, sizeof(SCreateAcctMsg));
SToken *pName = &pInfo->pMiscInfo->user.user;
SToken *pPwd = &pInfo->pMiscInfo->user.passwd;
......@@ -67,17 +65,18 @@ SCreateAcctMsg* buildAcctManipulationMsg(SSqlInfo* pInfo, int32_t* outputLen, in
}
*outputLen = sizeof(SCreateAcctMsg);
return pMsg;
return pCreateMsg;
}
SDropUserMsg* buildDropUserMsg(SSqlInfo* pInfo, int32_t *msgLen, int64_t id, char* msgBuf, int32_t msgBufLen) {
SToken* pName = taosArrayGet(pInfo->pMiscInfo->a, 0);
if (pName->n >= TSDB_USER_LEN) {
return NULL;
}
SDropUserMsg* pMsg = calloc(1, sizeof(SDropUserMsg));
if (pMsg == NULL) {
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return NULL;
}
......@@ -90,7 +89,6 @@ SShowMsg* buildShowMsg(SShowInfo* pShowInfo, SParseBasicCtx *pCtx, char* msgBuf,
SShowMsg* pShowMsg = calloc(1, sizeof(SShowMsg));
pShowMsg->type = pShowInfo->showType;
if (pShowInfo->showType != TSDB_MGMT_TABLE_VNODES) {
SToken* pPattern = &pShowInfo->pattern;
if (pPattern->type > 0) { // only show tables support wildcard query
......@@ -339,7 +337,7 @@ SDropStbMsg* buildDropStableMsg(SSqlInfo* pInfo, int32_t* len, SParseBasicCtx* p
SCreateDnodeMsg *buildCreateDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMsgBuf) {
const char* msg1 = "invalid host name (name too long, maximum length 128)";
const char* msg2 = "dnode name can not be string";
const char* msg3 = "port should be an integer that is less than 65535";
const char* msg3 = "port should be an integer that is less than 65535 and greater than 0";
const char* msg4 = "failed prepare create dnode message";
if (taosArrayGetSize(pInfo->pMiscInfo->a) != 2) {
......@@ -363,7 +361,7 @@ SCreateDnodeMsg *buildCreateDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMs
int64_t val = 0;
toInteger(port->z, port->n, 10, &val, &isSign);
if (val >= UINT16_MAX) {
if (val >= UINT16_MAX || val <= 0) {
buildInvalidOperationMsg(pMsgBuf, msg3);
return NULL;
}
......@@ -384,7 +382,6 @@ SCreateDnodeMsg *buildCreateDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMs
SDropDnodeMsg *buildDropDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMsgBuf) {
SToken* pzName = taosArrayGet(pInfo->pMiscInfo->a, 0);
char* end = NULL;
SDropDnodeMsg * pDrop = (SDropDnodeMsg *)calloc(1, sizeof(SDropDnodeMsg));
pDrop->dnodeId = strtoll(pzName->z, &end, 10);
......@@ -398,5 +395,4 @@ SDropDnodeMsg *buildDropDnodeMsg(SSqlInfo* pInfo, int32_t* len, SMsgBuf* pMsgBuf
}
return pDrop;
}
}
\ No newline at end of file
source/libs/parser/src/dCDAstProcess.c
浏览文件 @ 60b56c66
......@@ -18,7 +18,7 @@ static bool has(SArray* pFieldList, int32_t startIndex, const char* name) {
}
static int32_t setShowInfo(SShowInfo* pShowInfo, SParseBasicCtx* pCtx, void** output, int32_t* outputLen,
SMsgBuf* pMsgBuf) {
SEpSet* pEpSet, void** pExtension, SMsgBuf* pMsgBuf) {
const char* msg1 = "invalid name";
const char* msg2 = "wildcard string should be less than %d characters";
const char* msg3 = "database name too long";
......@@ -31,57 +31,87 @@ static int32_t setShowInfo(SShowInfo* pShowInfo, SParseBasicCtx* pCtx, void** ou
* wildcard in like clause in pInfo->pMiscInfo->a[1]
*/
int16_t showType = pShowInfo->showType;
if (showType == TSDB_MGMT_TABLE_STB || showType == TSDB_MGMT_TABLE_VGROUP) {
SToken* pDbPrefixToken = &pShowInfo->prefix;
if (pDbPrefixToken->type != 0) {
if (pDbPrefixToken->n >= TSDB_DB_NAME_LEN) { // db name is too long
return buildInvalidOperationMsg(pMsgBuf, msg3);
}
if (showType == TSDB_MGMT_TABLE_TABLE) {
SVShowTablesReq* pShowReq = calloc(1, sizeof(SVShowTablesReq));
if (pDbPrefixToken->n <= 0) {
return buildInvalidOperationMsg(pMsgBuf, msg5);
}
SArray* array = NULL;
SName name = {0};
tNameSetDbName(&name, pCtx->acctId, pCtx->db, strlen(pCtx->db));
if (parserValidateIdToken(pDbPrefixToken) != TSDB_CODE_SUCCESS) {
return buildInvalidOperationMsg(pMsgBuf, msg1);
}
char dbFname[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(&name, dbFname);
catalogGetDBVgroup(pCtx->pCatalog, pCtx->pTransporter, &pCtx->mgmtEpSet, dbFname, 0, &array);
// int32_t ret = tNameSetDbName(&pTableMetaInfo->name, getAccountId(pRequest->pTsc), pDbPrefixToken);
// if (ret != TSDB_CODE_SUCCESS) {
// return buildInvalidOperationMsg(pMsgBuf, msg1);
// }
SVgroupInfo* info = taosArrayGet(array, 0);
pShowReq->head.vgId = htonl(info->vgId);
pEpSet->numOfEps = info->numOfEps;
pEpSet->inUse = info->inUse;
for(int32_t i = 0; i < pEpSet->numOfEps; ++i) {
strncpy(pEpSet->fqdn[i], info->epAddr[i].fqdn, tListLen(pEpSet->fqdn[i]));
pEpSet->port[i] = info->epAddr[i].port;
}
// show table/stable like 'xxxx', set the like pattern for show tables
SToken* pPattern = &pShowInfo->pattern;
if (pPattern->type != 0) {
if (pPattern->type == TK_ID && pPattern->z[0] == TS_ESCAPE_CHAR) {
return buildInvalidOperationMsg(pMsgBuf, msg4);
*outputLen = sizeof(SVShowTablesReq);
*output = pShowReq;
*pExtension = array;
} else {
if (showType == TSDB_MGMT_TABLE_STB || showType == TSDB_MGMT_TABLE_VGROUP) {
SToken* pDbPrefixToken = &pShowInfo->prefix;
if (pDbPrefixToken->type != 0) {
if (pDbPrefixToken->n >= TSDB_DB_NAME_LEN) { // db name is too long
return buildInvalidOperationMsg(pMsgBuf, msg3);
}
if (pDbPrefixToken->n <= 0) {
return buildInvalidOperationMsg(pMsgBuf, msg5);
}
if (parserValidateIdToken(pDbPrefixToken) != TSDB_CODE_SUCCESS) {
return buildInvalidOperationMsg(pMsgBuf, msg1);
}
// int32_t ret = tNameSetDbName(&pTableMetaInfo->name, getAccountId(pRequest->pTsc), pDbPrefixToken);
// if (ret != TSDB_CODE_SUCCESS) {
// return buildInvalidOperationMsg(pMsgBuf, msg1);
// }
}
pPattern->n = strdequote(pPattern->z);
if (pPattern->n <= 0) {
return buildInvalidOperationMsg(pMsgBuf, msg6);
// show table/stable like 'xxxx', set the like pattern for show tables
SToken* pPattern = &pShowInfo->pattern;
if (pPattern->type != 0) {
if (pPattern->type == TK_ID && pPattern->z[0] == TS_ESCAPE_CHAR) {
return buildInvalidOperationMsg(pMsgBuf, msg4);
}
pPattern->n = strdequote(pPattern->z);
if (pPattern->n <= 0) {
return buildInvalidOperationMsg(pMsgBuf, msg6);
}
if (pPattern->n > tsMaxWildCardsLen) {
char tmp[64] = {0};
sprintf(tmp, msg2, tsMaxWildCardsLen);
return buildInvalidOperationMsg(pMsgBuf, tmp);
}
}
} else if (showType == TSDB_MGMT_TABLE_VNODES) {
if (pShowInfo->prefix.type == 0) {
return buildInvalidOperationMsg(pMsgBuf, "No specified dnode ep");
}
if (pPattern->n > tsMaxWildCardsLen) {
char tmp[64] = {0};
sprintf(tmp, msg2, tsMaxWildCardsLen);
return buildInvalidOperationMsg(pMsgBuf, tmp);
if (pShowInfo->prefix.type == TK_STRING) {
pShowInfo->prefix.n = strdequote(pShowInfo->prefix.z);
}
}
} else if (showType == TSDB_MGMT_TABLE_VNODES) {
if (pShowInfo->prefix.type == 0) {
return buildInvalidOperationMsg(pMsgBuf, "No specified dnode ep");
}
if (pShowInfo->prefix.type == TK_STRING) {
pShowInfo->prefix.n = strdequote(pShowInfo->prefix.z);
}
*pEpSet = pCtx->mgmtEpSet;
*output = buildShowMsg(pShowInfo, pCtx, pMsgBuf->buf, pMsgBuf->len);
*outputLen = sizeof(SShowMsg) /* + htons(pShowMsg->payloadLen)*/;
}
*output = buildShowMsg(pShowInfo, pCtx, pMsgBuf->buf, pMsgBuf->len);
*outputLen = sizeof(SShowMsg) /* + htons(pShowMsg->payloadLen)*/;
return TSDB_CODE_SUCCESS;
}
......@@ -608,8 +638,9 @@ int32_t qParserValidateDclSqlNode(SSqlInfo* pInfo, SParseBasicCtx* pCtx, SDclStm
}
case TSDB_SQL_SHOW: {
code = setShowInfo(&pInfo->pMiscInfo->showOpt, pCtx, (void**)&pDcl->pMsg, &pDcl->msgLen, pMsgBuf);
pDcl->msgType = TDMT_MND_SHOW;
SShowInfo* pShowInfo = &pInfo->pMiscInfo->showOpt;
code = setShowInfo(pShowInfo, pCtx, (void**)&pDcl->pMsg, &pDcl->msgLen, &pDcl->epSet, &pDcl->pExtension, pMsgBuf);
pDcl->msgType = (pShowInfo->showType == TSDB_MGMT_TABLE_TABLE)? TDMT_VND_SHOW_TABLES:TDMT_MND_SHOW;
break;
}
......
source/libs/parser/src/insertParser.c
浏览文件 @ 60b56c66
......@@ -64,64 +64,6 @@ typedef struct SInsertParseContext {
SInsertStmtInfo* pOutput;
} SInsertParseContext;
static FORCE_INLINE int32_t toDouble(SToken *pToken, double *value, char **endPtr) {
errno = 0;
*value = strtold(pToken->z, endPtr);
// not a valid integer number, return error
if ((*endPtr - pToken->z) != pToken->n) {
return TK_ILLEGAL;
}
return pToken->type;
}
static int32_t toInt64(const char* z, int16_t type, int32_t n, int64_t* value, bool issigned) {
errno = 0;
int32_t ret = 0;
char* endPtr = NULL;
if (type == TK_FLOAT) {
double v = strtod(z, &endPtr);
if ((errno == ERANGE && v == HUGE_VALF) || isinf(v) || isnan(v)) {
ret = -1;
} else if ((issigned && (v < INT64_MIN || v > INT64_MAX)) || ((!issigned) && (v < 0 || v > UINT64_MAX))) {
ret = -1;
} else {
*value = (int64_t) round(v);
}
errno = 0;
return ret;
}
int32_t radix = 10;
if (type == TK_HEX) {
radix = 16;
} else if (type == TK_BIN) {
radix = 2;
}
// the string may be overflow according to errno
if (!issigned) {
const char *p = z;
while(*p != 0 && *p == ' ') p++;
if (*p != 0 && *p == '-') { return -1;}
*value = strtoull(z, &endPtr, radix);
} else {
*value = strtoll(z, &endPtr, radix);
}
// not a valid integer number, return error
if (endPtr - z != n || errno == ERANGE) {
ret = -1;
}
errno = 0;
return ret;
}
static int32_t skipInsertInto(SInsertParseContext* pCxt) {
SToken sToken;
NEXT_TOKEN(pCxt->pSql, sToken);
......@@ -159,10 +101,8 @@ static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) {
char tableName[TSDB_TABLE_FNAME_LEN] = {0};
tNameExtractFullName(&name, tableName);
SParseBasicCtx* pBasicCtx = &pCxt->pComCxt->ctx;
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)));
......@@ -349,207 +289,6 @@ static FORCE_INLINE int32_t MemRowAppend(const void *value, int32_t len, void *p
return TSDB_CODE_SUCCESS;
}
//static FORCE_INLINE int32_t checkAndTrimValue(SToken* pToken, uint32_t type, char* tmpTokenBuf, SMsgBuf* pMsgBuf) {
// if ((type != TK_NOW && type != TK_INTEGER && type != TK_STRING && type != TK_FLOAT && type != TK_BOOL &&
// type != TK_NULL && type != TK_HEX && type != TK_OCT && type != TK_BIN) ||
// (pToken->n == 0) || (type == TK_RP)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid data or symbol", pToken->z);
// }
//
// if (IS_NUMERIC_TYPE(type) && pToken->n == 0) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid numeric data", pToken->z);
// }
//
// // Remove quotation marks
// if (TK_STRING == type) {
// if (pToken->n >= TSDB_MAX_BYTES_PER_ROW) {
// return buildSyntaxErrMsg(pMsgBuf, "too long string", pToken->z);
// }
//
// // delete escape character: \\, \', \"
// char delim = pToken->z[0];
// int32_t cnt = 0;
// int32_t j = 0;
// for (uint32_t k = 1; k < pToken->n - 1; ++k) {
// if (pToken->z[k] == '\\' || (pToken->z[k] == delim && pToken->z[k + 1] == delim)) {
// tmpTokenBuf[j] = pToken->z[k + 1];
// cnt++;
// j++;
// k++;
// continue;
// }
// tmpTokenBuf[j] = pToken->z[k];
// j++;
// }
//
// tmpTokenBuf[j] = 0;
// pToken->z = tmpTokenBuf;
// pToken->n -= 2 + cnt;
// }
//
// return TSDB_CODE_SUCCESS;
//}
//static FORCE_INLINE int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t timePrec, char* tmpTokenBuf, _row_append_fn_t func, void* param, SMsgBuf* pMsgBuf) {
// int64_t iv;
// char *endptr = NULL;
// bool isSigned = false;
//
// CHECK_CODE(checkAndTrimValue(pToken, pSchema->type, tmpTokenBuf, pMsgBuf));
//
// if (isNullStr(pToken)) {
// if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
// int64_t tmpVal = 0;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// return func(getNullValue(pSchema->type), 0, param);
// }
//
// switch (pSchema->type) {
// case TSDB_DATA_TYPE_BOOL: {
// if ((pToken->type == TK_BOOL || pToken->type == TK_STRING) && (pToken->n != 0)) {
// if (strncmp(pToken->z, "true", pToken->n) == 0) {
// return func(&TRUE_VALUE, pSchema->bytes, param);
// } else if (strncmp(pToken->z, "false", pToken->n) == 0) {
// return func(&FALSE_VALUE, pSchema->bytes, param);
// } else {
// return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
// }
// } else if (pToken->type == TK_INTEGER) {
// return func(((strtoll(pToken->z, NULL, 10) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes, param);
// } else if (pToken->type == TK_FLOAT) {
// return func(((strtod(pToken->z, NULL) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes, param);
// } else {
// return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
// }
// }
//
// case TSDB_DATA_TYPE_TINYINT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid tinyint data", pToken->z);
// } else if (!IS_VALID_TINYINT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "tinyint data overflow", pToken->z);
// }
//
// uint8_t tmpVal = (uint8_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_UTINYINT:{
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned tinyint data", pToken->z);
// } else if (!IS_VALID_UTINYINT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "unsigned tinyint data overflow", pToken->z);
// }
// uint8_t tmpVal = (uint8_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_SMALLINT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid smallint data", pToken->z);
// } else if (!IS_VALID_SMALLINT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "smallint data overflow", pToken->z);
// }
// int16_t tmpVal = (int16_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_USMALLINT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned smallint data", pToken->z);
// } else if (!IS_VALID_USMALLINT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "unsigned smallint data overflow", pToken->z);
// }
// uint16_t tmpVal = (uint16_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_INT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid int data", pToken->z);
// } else if (!IS_VALID_INT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "int data overflow", pToken->z);
// }
// int32_t tmpVal = (int32_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_UINT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned int data", pToken->z);
// } else if (!IS_VALID_UINT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "unsigned int data overflow", pToken->z);
// }
// uint32_t tmpVal = (uint32_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_BIGINT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid bigint data", pToken->z);
// } else if (!IS_VALID_BIGINT(iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "bigint data overflow", pToken->z);
// }
// return func(&iv, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_UBIGINT: {
// if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned bigint data", pToken->z);
// } else if (!IS_VALID_UBIGINT((uint64_t)iv)) {
// return buildSyntaxErrMsg(pMsgBuf, "unsigned bigint data overflow", pToken->z);
// }
// uint64_t tmpVal = (uint64_t)iv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_FLOAT: {
// double dv;
// if (TK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
// return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
// }
// if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || dv > FLT_MAX || dv < -FLT_MAX || isinf(dv) || isnan(dv)) {
// return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
// }
// float tmpVal = (float)dv;
// return func(&tmpVal, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_DOUBLE: {
// double dv;
// if (TK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
// return buildSyntaxErrMsg(pMsgBuf, "illegal double data", pToken->z);
// }
// if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || isinf(dv) || isnan(dv)) {
// return buildSyntaxErrMsg(pMsgBuf, "illegal double data", pToken->z);
// }
// return func(&dv, pSchema->bytes, param);
// }
//
// case TSDB_DATA_TYPE_BINARY: {
// // too long values will return invalid sql, not be truncated automatically
// if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) {
// return buildSyntaxErrMsg(pMsgBuf, "string data overflow", pToken->z);
// }
// return func(pToken->z, pToken->n, param);
// }
// case TSDB_DATA_TYPE_NCHAR: {
// return func(pToken->z, pToken->n, param);
// }
// case TSDB_DATA_TYPE_TIMESTAMP: {
// int64_t tmpVal;
// if (parseTime(end, pToken, timePrec, &tmpVal, pMsgBuf) != TSDB_CODE_SUCCESS) {
// return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp", pToken->z);
// }
// return func(&tmpVal, pSchema->bytes, param);
// }
// }
//
// return TSDB_CODE_FAILED;
//}
// pSql -> tag1_name, ...)
static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo* pColList, SSchema* pSchema) {
int32_t nCols = pColList->numOfCols;
......
source/libs/parser/src/parser.c
浏览文件 @ 60b56c66
......@@ -58,13 +58,11 @@ int32_t parseQuerySql(SParseContext* pCxt, SQueryNode** pQuery) {
} else {
SQueryStmtInfo* pQueryInfo = calloc(1, sizeof(SQueryStmtInfo));
if (pQueryInfo == NULL) {
terrno = TSDB_CODE_TSC_OUT_OF_MEMORY; // set correct error code.
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY; // set correct error code.
return terrno;
}
struct SCatalog* pCatalog = NULL;
int32_t code = catalogGetHandle(NULL, &pCatalog);
code = qParserValidateSqlNode(pCatalog, &info, pQueryInfo, pCxt->ctx.requestId, pCxt->pMsg, pCxt->msgLen);
int32_t code = qParserValidateSqlNode(pCxt->ctx.pCatalog, &info, pQueryInfo, pCxt->ctx.requestId, pCxt->pMsg, pCxt->msgLen);
if (code == TSDB_CODE_SUCCESS) {
*pQuery = (SQueryNode*)pQueryInfo;
}
......@@ -220,7 +218,7 @@ int32_t qParserExtractRequestedMetaInfo(const SSqlInfo* pSqlInfo, SCatalogReq* p
return code;
}
void qParserClearupMetaRequestInfo(SCatalogReq* pMetaReq) {
void qParserCleanupMetaRequestInfo(SCatalogReq* pMetaReq) {
if (pMetaReq == NULL) {
return;
}
......
source/libs/parser/src/parserUtil.c
浏览文件 @ 60b56c66
......@@ -563,16 +563,6 @@ TAOS_FIELD createField(const SSchema* pSchema) {
return f;
}
SSchema createSchema(uint8_t type, int16_t bytes, int16_t colId, const char* name) {
SSchema s = {0};
s.type = type;
s.bytes = bytes;
s.colId = colId;
tstrncpy(s.name, name, tListLen(s.name));
return s;
}
void setColumn(SColumn* pColumn, uint64_t uid, const char* tableName, int8_t flag, const SSchema* pSchema) {
pColumn->uid = uid;
pColumn->flag = flag;
......@@ -1649,9 +1639,9 @@ static bool isNullStr(SToken *pToken) {
}
static FORCE_INLINE int32_t checkAndTrimValue(SToken* pToken, uint32_t type, char* tmpTokenBuf, SMsgBuf* pMsgBuf) {
if ((type != TK_NOW && type != TK_INTEGER && type != TK_STRING && type != TK_FLOAT && type != TK_BOOL &&
type != TK_NULL && type != TK_HEX && type != TK_OCT && type != TK_BIN) ||
(pToken->n == 0) || (type == TK_RP)) {
if ((pToken->type != TK_NOW && pToken->type != TK_INTEGER && pToken->type != TK_STRING && pToken->type != TK_FLOAT && pToken->type != TK_BOOL &&
pToken->type != TK_NULL && pToken->type != TK_HEX && pToken->type != TK_OCT && pToken->type != TK_BIN) ||
(pToken->n == 0) || (pToken->type == TK_RP)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid data or symbol", pToken->z);
}
......@@ -1795,7 +1785,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_TINYINT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid tinyint data", pToken->z);
} else if (!IS_VALID_TINYINT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "tinyint data overflow", pToken->z);
......@@ -1806,7 +1796,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_UTINYINT:{
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned tinyint data", pToken->z);
} else if (!IS_VALID_UTINYINT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "unsigned tinyint data overflow", pToken->z);
......@@ -1816,7 +1806,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_SMALLINT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid smallint data", pToken->z);
} else if (!IS_VALID_SMALLINT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "smallint data overflow", pToken->z);
......@@ -1826,7 +1816,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_USMALLINT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned smallint data", pToken->z);
} else if (!IS_VALID_USMALLINT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "unsigned smallint data overflow", pToken->z);
......@@ -1836,7 +1826,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_INT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid int data", pToken->z);
} else if (!IS_VALID_INT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "int data overflow", pToken->z);
......@@ -1846,7 +1836,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_UINT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned int data", pToken->z);
} else if (!IS_VALID_UINT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "unsigned int data overflow", pToken->z);
......@@ -1856,7 +1846,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_BIGINT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid bigint data", pToken->z);
} else if (!IS_VALID_BIGINT(iv)) {
return buildSyntaxErrMsg(pMsgBuf, "bigint data overflow", pToken->z);
......@@ -1865,7 +1855,7 @@ int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t ti
}
case TSDB_DATA_TYPE_UBIGINT: {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, pToken->type, &iv, &isSigned)) {
if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned bigint data", pToken->z);
} else if (!IS_VALID_UBIGINT((uint64_t)iv)) {
return buildSyntaxErrMsg(pMsgBuf, "unsigned bigint data overflow", pToken->z);
......
source/libs/parser/test/mockCatalogService.cpp
浏览文件 @ 60b56c66
......@@ -97,8 +97,8 @@ public:
int32_t catalogGetTableMeta(const SName* pTableName, STableMeta** pTableMeta) const {
std::unique_ptr<STableMeta> table;
char db[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(pTableName, db);
char db[TSDB_DB_NAME_LEN] = {0};
tNameGetDbName(pTableName, db);
const char* tname = tNameGetTableName(pTableName);
int32_t code = copyTableSchemaMeta(db, tname, &table);
......@@ -111,6 +111,7 @@ public:
int32_t catalogGetTableHashVgroup(const SName* pTableName, SVgroupInfo* vgInfo) const {
// todo
vgInfo->vgId = 1;
return 0;
}
......
source/libs/parser/test/parserTests.cpp
浏览文件 @ 60b56c66
......@@ -99,7 +99,7 @@ void sqlCheck(const char* sql, bool valid) {
}
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -159,7 +159,7 @@ TEST(testCase, validateAST_test) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 3);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -205,7 +205,7 @@ TEST(testCase, function_Test) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 1);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -251,7 +251,7 @@ TEST(testCase, function_Test2) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 1);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -296,7 +296,7 @@ TEST(testCase, function_Test3) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 6);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -342,7 +342,7 @@ TEST(testCase, function_Test4) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 1);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -393,7 +393,7 @@ TEST(testCase, function_Test5) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 1);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -507,7 +507,7 @@ TEST(testCase, function_Test6) {
ASSERT_STREQ(p2->pExpr->_function.pChild[0]->pSchema->name, "t.1abc.b*a");
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -569,7 +569,7 @@ TEST(testCase, function_Test6) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, numOfCols);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -625,7 +625,7 @@ TEST(testCase, function_Test6) {
ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
......@@ -648,7 +648,7 @@ TEST(testCase, function_Test6) {
ASSERT_EQ(ret, 0);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......@@ -678,7 +678,7 @@ TEST(testCase, function_Test6) {
ASSERT_NE(ret, 0);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
//===============================================================================================================
info1 = doGenerateAST("select top(a*b, ABC) from `t.1abc` interval(10s, 1s)");
......@@ -700,7 +700,7 @@ TEST(testCase, function_Test6) {
ASSERT_NE(ret, 0);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......
source/libs/parser/test/plannerTest.cpp
浏览文件 @ 60b56c66
......@@ -102,7 +102,7 @@ void generateLogicplan(const char* sql) {
printf("%s\n", str);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
}
......@@ -163,7 +163,7 @@ TEST(testCase, planner_test) {
printf("%s\n", str);
destroyQueryInfo(pQueryInfo);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
......
source/libs/parser/test/tokenizerTest.cpp
浏览文件 @ 60b56c66
......@@ -714,7 +714,7 @@ TEST(testCase, extractMeta_test) {
ASSERT_EQ(ret, 0);
ASSERT_EQ(taosArrayGetSize(req.pTableName), 1);
qParserClearupMetaRequestInfo(&req);
qParserCleanupMetaRequestInfo(&req);
destroySqlInfo(&info1);
}
source/libs/planner/src/physicalPlan.c
浏览文件 @ 60b56c66
......@@ -207,6 +207,7 @@ static SSubplan* initSubplan(SPlanContext* pCxt, int32_t type) {
}
taosArrayPush(currentLevel, &subplan);
pCxt->pCurrentSubplan = subplan;
++(pCxt->pDag->numOfSubplans);
return subplan;
}
......@@ -293,11 +294,14 @@ static void splitInsertSubplan(SPlanContext* pCxt, SQueryPlanNode* pPlanNode) {
SArray* vgs = (SArray*)pPlanNode->pExtInfo;
size_t numOfVg = taosArrayGetSize(vgs);
for (int32_t i = 0; i < numOfVg; ++i) {
STORE_CURRENT_SUBPLAN(pCxt);
SSubplan* subplan = initSubplan(pCxt, QUERY_TYPE_MODIFY);
SVgDataBlocks* blocks = (SVgDataBlocks*)taosArrayGetP(vgs, i);
vgroupInfoToEpSet(&blocks->vg, &subplan->execEpSet);
subplan->pNode = NULL;
subplan->pDataSink = createDataInserter(pCxt, blocks);
subplan->type = QUERY_TYPE_MODIFY;
RECOVERY_CURRENT_SUBPLAN(pCxt);
}
}
......
source/libs/planner/test/plannerTests.cpp
浏览文件 @ 60b56c66
......@@ -100,6 +100,6 @@ TEST(testCase, planner_test) {
// ASSERT_EQ(pQueryInfo->fieldsInfo.numOfOutput, 2);
//
// destroyQueryInfo(pQueryInfo);
// qParserClearupMetaRequestInfo(&req);
// qParserCleanupMetaRequestInfo(&req);
// destroySqlInfo(&info1);
}
\ No newline at end of file
source/libs/qworker/src/qworker.c
浏览文件 @ 60b56c66
#include "tmsg.h"
#include "query.h"
#include "qworker.h"
#include "qworkerInt.h"
#include "tname.h"
#include "planner.h"
#include "query.h"
#include "qworkerInt.h"
#include "tmsg.h"
int32_t qwValidateStatus(int8_t oriStatus, int8_t newStatus) {
int32_t code = 0;
......@@ -136,7 +137,6 @@ static int32_t qwAddScheduler(int32_t rwType, SQWorkerMgmt *mgmt, uint64_t sId,
return TSDB_CODE_SUCCESS;
}
static int32_t qwAcquireScheduler(int32_t rwType, SQWorkerMgmt *mgmt, uint64_t sId, SQWSchStatus **sch, int32_t nOpt) {
QW_LOCK(rwType, &mgmt->schLock);
*sch = taosHashGet(mgmt->schHash, &sId, sizeof(sId));
......@@ -155,8 +155,6 @@ static int32_t qwAcquireScheduler(int32_t rwType, SQWorkerMgmt *mgmt, uint64_t s
return TSDB_CODE_SUCCESS;
}
static FORCE_INLINE void qwReleaseScheduler(int32_t rwType, SQWorkerMgmt *mgmt) {
QW_UNLOCK(rwType, &mgmt->schLock);
}
......@@ -180,12 +178,10 @@ static int32_t qwAcquireTask(int32_t rwType, SQWSchStatus *sch, uint64_t qId, ui
return qwAcquireTaskImpl(rwType, sch, qId, tId, task);
}
static FORCE_INLINE void qwReleaseTask(int32_t rwType, SQWSchStatus *sch) {
QW_UNLOCK(rwType, &sch->tasksLock);
}
int32_t qwAddTaskToSch(int32_t rwType, SQWSchStatus *sch, uint64_t qId, uint64_t tId, int8_t status, int32_t eOpt, SQWTaskStatus **task) {
int32_t code = 0;
......@@ -232,7 +228,6 @@ int32_t qwAddTaskToSch(int32_t rwType, SQWSchStatus *sch, uint64_t qId, uint64_t
return TSDB_CODE_SUCCESS;
}
static int32_t qwAddTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int32_t status, int32_t eOpt, SQWSchStatus **sch, SQWTaskStatus **task) {
SQWSchStatus *tsch = NULL;
QW_ERR_RET(qwAcquireScheduler(QW_READ, mgmt, sId, &tsch, QW_NOT_EXIST_ADD));
......@@ -251,8 +246,6 @@ static int32_t qwAddTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_
QW_RET(code);
}
static FORCE_INLINE int32_t qwAcquireTaskResCache(int32_t rwType, SQWorkerMgmt *mgmt, uint64_t queryId, uint64_t taskId, SQWorkerResCache **res) {
char id[sizeof(queryId) + sizeof(taskId)] = {0};
QW_SET_QTID(id, queryId, taskId);
......@@ -444,8 +437,6 @@ _return:
QW_RET(code);
}
int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t queryId, uint64_t taskId) {
SQWSchStatus *sch = NULL;
SQWTaskStatus *task = NULL;
......@@ -479,7 +470,6 @@ int32_t qwDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t queryId, uint64_t
return TSDB_CODE_SUCCESS;
}
int32_t qwCancelDropTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t queryId, uint64_t taskId) {
SQWSchStatus *sch = NULL;
SQWTaskStatus *task = NULL;
......@@ -547,8 +537,6 @@ _return:
QW_RET(code);
}
int32_t qwBuildAndSendQueryRsp(SRpcMsg *pMsg, int32_t code) {
SQueryTableRsp *pRsp = (SQueryTableRsp *)rpcMallocCont(sizeof(SQueryTableRsp));
pRsp->code = code;
......@@ -634,7 +622,6 @@ int32_t qwBuildAndSendFetchRsp(SRpcMsg *pMsg, void *data) {
return TSDB_CODE_SUCCESS;
}
int32_t qwBuildAndSendCancelRsp(SRpcMsg *pMsg, int32_t code) {
STaskCancelRsp *pRsp = (STaskCancelRsp *)rpcMallocCont(sizeof(STaskCancelRsp));
pRsp->code = code;
......@@ -648,7 +635,6 @@ int32_t qwBuildAndSendCancelRsp(SRpcMsg *pMsg, int32_t code) {
};
rpcSendResponse(&rpcRsp);
return TSDB_CODE_SUCCESS;
}
......@@ -665,11 +651,71 @@ int32_t qwBuildAndSendDropRsp(SRpcMsg *pMsg, int32_t code) {
};
rpcSendResponse(&rpcRsp);
return TSDB_CODE_SUCCESS;
}
int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
int32_t numOfCols = 6;
int32_t msgSize = sizeof(SVShowTablesRsp) + sizeof(SSchema) * numOfCols;
SVShowTablesRsp *pRsp = (SVShowTablesRsp *)rpcMallocCont(msgSize);
int32_t cols = 0;
SSchema *pSchema = pRsp->metaInfo.pSchema;
const SSchema *s = tGetTbnameColumnSchema();
*pSchema = createSchema(s->type, htonl(s->bytes), htonl(++cols), "name");
pSchema++;
int32_t type = TSDB_DATA_TYPE_TIMESTAMP;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "created");
pSchema++;
type = TSDB_DATA_TYPE_SMALLINT;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "columns");
pSchema++;
*pSchema = createSchema(s->type, htonl(s->bytes), htonl(++cols), "stable");
pSchema++;
type = TSDB_DATA_TYPE_BIGINT;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "uid");
pSchema++;
type = TSDB_DATA_TYPE_INT;
*pSchema = createSchema(type, htonl(tDataTypes[type].bytes), htonl(++cols), "vgId");
assert(cols == numOfCols);
pRsp->metaInfo.numOfColumns = htonl(cols);
SRpcMsg rpcMsg = {
.handle = pMsg->handle,
.ahandle = pMsg->ahandle,
.pCont = pRsp,
.contLen = msgSize,
.code = code,
};
rpcSendResponse(&rpcMsg);
return TSDB_CODE_SUCCESS;
}
int32_t qwBuildAndSendShowFetchRsp(SRpcMsg *pMsg, SVShowTablesFetchReq* pFetchReq) {
SVShowTablesFetchRsp *pRsp = (SVShowTablesFetchRsp *)rpcMallocCont(sizeof(SVShowTablesFetchRsp));
int32_t handle = htonl(pFetchReq->id);
pRsp->numOfRows = 0;
SRpcMsg rpcMsg = {
.handle = pMsg->handle,
.ahandle = pMsg->ahandle,
.pCont = pRsp,
.contLen = sizeof(*pRsp),
.code = 0,
};
rpcSendResponse(&rpcMsg);
return TSDB_CODE_SUCCESS;
}
int32_t qwCheckAndSendReadyRsp(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t queryId, uint64_t taskId, SRpcMsg *pMsg, int32_t rspCode) {
SQWSchStatus *sch = NULL;
......@@ -801,7 +847,6 @@ int32_t qwCheckTaskCancelDrop( SQWorkerMgmt *mgmt, uint64_t sId, uint64_t queryI
return TSDB_CODE_SUCCESS;
}
int32_t qwHandleFetch(SQWorkerResCache *res, SQWorkerMgmt *mgmt, uint64_t sId, uint64_t queryId, uint64_t taskId, SRpcMsg *pMsg) {
SQWSchStatus *sch = NULL;
SQWTaskStatus *task = NULL;
......@@ -911,7 +956,6 @@ int32_t qwQueryPostProcess(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint6
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerInit(SQWorkerCfg *cfg, void **qWorkerMgmt) {
SQWorkerMgmt *mgmt = calloc(1, sizeof(SQWorkerMgmt));
if (NULL == mgmt) {
......@@ -1157,6 +1201,25 @@ _return:
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerProcessShowMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT;
}
int32_t code = 0;
SVShowTablesReq *pReq = pMsg->pCont;
QW_ERR_RET(qwBuildAndSendShowRsp(pMsg, code));
}
int32_t qWorkerProcessShowFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT;
}
SVShowTablesFetchReq *pFetchReq = pMsg->pCont;
QW_ERR_RET(qwBuildAndSendShowFetchRsp(pMsg, pFetchReq));
}
int32_t qWorkerContinueQuery(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
int32_t code = 0;
int8_t status = 0;
......@@ -1182,7 +1245,6 @@ int32_t qWorkerContinueQuery(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
QW_RET(code);
}
void qWorkerDestroy(void **qWorkerMgmt) {
if (NULL == qWorkerMgmt || NULL == *qWorkerMgmt) {
return;
......
source/libs/scheduler/src/scheduler.c
浏览文件 @ 60b56c66
......@@ -166,7 +166,7 @@ int32_t schValidateAndBuildJob(SQueryDag *dag, SSchJob *job) {
}
for (int32_t n = 0; n < levelPlanNum; ++n) {
SSubplan *plan = taosArrayGet(levelPlans, n);
SSubplan *plan = taosArrayGetP(levelPlans, n);
SSchTask task = {0};
if (plan->type == QUERY_TYPE_MODIFY) {
......
source/util/test/hashTest.cpp
浏览文件 @ 60b56c66
......@@ -154,9 +154,9 @@ void acquireRleaseTest() {
int32_t code = 0;
int32_t num = 0;
TESTSTRUCT data = {0};
char *str1 = "abcdefg";
char *str2 = "aaaaaaa";
char *str3 = "123456789";
const char *str1 = "abcdefg";
const char *str2 = "aaaaaaa";
const char *str3 = "123456789";
data.p = (char *)malloc(10);
strcpy(data.p, str1);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或