Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 c67e5722 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/tsdb_refact

上级 27158d69 63e78e3e
展开全部 显示空白变更内容
内联 并排
Showing with 4791 addition and 4188 deletion
include/common/tmsg.h
浏览文件 @ c67e5722
......@@ -1506,6 +1506,7 @@ typedef struct {
char* sql;
char* ast;
int8_t triggerType;
int64_t maxDelay;
int64_t watermark;
} SCMCreateStreamReq;
......
include/common/ttokendef.h
浏览文件 @ c67e5722
......@@ -184,76 +184,77 @@
#define TK_TRIGGER 166
#define TK_AT_ONCE 167
#define TK_WINDOW_CLOSE 168
#define TK_WATERMARK 169
#define TK_KILL 170
#define TK_CONNECTION 171
#define TK_TRANSACTION 172
#define TK_BALANCE 173
#define TK_VGROUP 174
#define TK_MERGE 175
#define TK_REDISTRIBUTE 176
#define TK_SPLIT 177
#define TK_SYNCDB 178
#define TK_DELETE 179
#define TK_NULL 180
#define TK_NK_QUESTION 181
#define TK_NK_ARROW 182
#define TK_ROWTS 183
#define TK_TBNAME 184
#define TK_QSTARTTS 185
#define TK_QENDTS 186
#define TK_WSTARTTS 187
#define TK_WENDTS 188
#define TK_WDURATION 189
#define TK_CAST 190
#define TK_NOW 191
#define TK_TODAY 192
#define TK_TIMEZONE 193
#define TK_COUNT 194
#define TK_FIRST 195
#define TK_LAST 196
#define TK_LAST_ROW 197
#define TK_BETWEEN 198
#define TK_IS 199
#define TK_NK_LT 200
#define TK_NK_GT 201
#define TK_NK_LE 202
#define TK_NK_GE 203
#define TK_NK_NE 204
#define TK_MATCH 205
#define TK_NMATCH 206
#define TK_CONTAINS 207
#define TK_JOIN 208
#define TK_INNER 209
#define TK_SELECT 210
#define TK_DISTINCT 211
#define TK_WHERE 212
#define TK_PARTITION 213
#define TK_BY 214
#define TK_SESSION 215
#define TK_STATE_WINDOW 216
#define TK_SLIDING 217
#define TK_FILL 218
#define TK_VALUE 219
#define TK_NONE 220
#define TK_PREV 221
#define TK_LINEAR 222
#define TK_NEXT 223
#define TK_HAVING 224
#define TK_ORDER 225
#define TK_SLIMIT 226
#define TK_SOFFSET 227
#define TK_LIMIT 228
#define TK_OFFSET 229
#define TK_ASC 230
#define TK_NULLS 231
#define TK_ID 232
#define TK_NK_BITNOT 233
#define TK_INSERT 234
#define TK_VALUES 235
#define TK_IMPORT 236
#define TK_NK_SEMI 237
#define TK_FILE 238
#define TK_MAX_DELAY 169
#define TK_WATERMARK 170
#define TK_KILL 171
#define TK_CONNECTION 172
#define TK_TRANSACTION 173
#define TK_BALANCE 174
#define TK_VGROUP 175
#define TK_MERGE 176
#define TK_REDISTRIBUTE 177
#define TK_SPLIT 178
#define TK_SYNCDB 179
#define TK_DELETE 180
#define TK_NULL 181
#define TK_NK_QUESTION 182
#define TK_NK_ARROW 183
#define TK_ROWTS 184
#define TK_TBNAME 185
#define TK_QSTARTTS 186
#define TK_QENDTS 187
#define TK_WSTARTTS 188
#define TK_WENDTS 189
#define TK_WDURATION 190
#define TK_CAST 191
#define TK_NOW 192
#define TK_TODAY 193
#define TK_TIMEZONE 194
#define TK_COUNT 195
#define TK_FIRST 196
#define TK_LAST 197
#define TK_LAST_ROW 198
#define TK_BETWEEN 199
#define TK_IS 200
#define TK_NK_LT 201
#define TK_NK_GT 202
#define TK_NK_LE 203
#define TK_NK_GE 204
#define TK_NK_NE 205
#define TK_MATCH 206
#define TK_NMATCH 207
#define TK_CONTAINS 208
#define TK_JOIN 209
#define TK_INNER 210
#define TK_SELECT 211
#define TK_DISTINCT 212
#define TK_WHERE 213
#define TK_PARTITION 214
#define TK_BY 215
#define TK_SESSION 216
#define TK_STATE_WINDOW 217
#define TK_SLIDING 218
#define TK_FILL 219
#define TK_VALUE 220
#define TK_NONE 221
#define TK_PREV 222
#define TK_LINEAR 223
#define TK_NEXT 224
#define TK_HAVING 225
#define TK_ORDER 226
#define TK_SLIMIT 227
#define TK_SOFFSET 228
#define TK_LIMIT 229
#define TK_OFFSET 230
#define TK_ASC 231
#define TK_NULLS 232
#define TK_ID 233
#define TK_NK_BITNOT 234
#define TK_INSERT 235
#define TK_VALUES 236
#define TK_IMPORT 237
#define TK_NK_SEMI 238
#define TK_FILE 239
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301
......
include/libs/function/functionMgt.h
浏览文件 @ c67e5722
......@@ -175,6 +175,7 @@ bool fmIsRepeatScanFunc(int32_t funcId);
bool fmIsUserDefinedFunc(int32_t funcId);
bool fmIsDistExecFunc(int32_t funcId);
bool fmIsForbidFillFunc(int32_t funcId);
bool fmIsForbidStreamFunc(int32_t funcId);
int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc, SFunctionNode** pMergeFunc);
......
include/libs/nodes/cmdnodes.h
浏览文件 @ c67e5722
......@@ -289,6 +289,7 @@ typedef struct SKillStmt {
typedef struct SStreamOptions {
ENodeType type;
int8_t triggerType;
SNode* pDelay;
SNode* pWatermark;
} SStreamOptions;
......
include/libs/nodes/nodes.h
浏览文件 @ c67e5722
......@@ -252,22 +252,20 @@ typedef struct SNodeList {
SListCell* pTail;
} SNodeList;
#define SNodeptr void*
SNodeptr nodesMakeNode(ENodeType type);
void nodesDestroyNode(SNodeptr pNode);
SNode* nodesMakeNode(ENodeType type);
void nodesDestroyNode(SNode* pNode);
SNodeList* nodesMakeList();
int32_t nodesListAppend(SNodeList* pList, SNodeptr pNode);
int32_t nodesListStrictAppend(SNodeList* pList, SNodeptr pNode);
int32_t nodesListMakeAppend(SNodeList** pList, SNodeptr pNode);
int32_t nodesListMakeStrictAppend(SNodeList** pList, SNodeptr pNode);
int32_t nodesListAppend(SNodeList* pList, SNode* pNode);
int32_t nodesListStrictAppend(SNodeList* pList, SNode* pNode);
int32_t nodesListMakeAppend(SNodeList** pList, SNode* pNode);
int32_t nodesListMakeStrictAppend(SNodeList** pList, SNode* pNode);
int32_t nodesListAppendList(SNodeList* pTarget, SNodeList* pSrc);
int32_t nodesListStrictAppendList(SNodeList* pTarget, SNodeList* pSrc);
int32_t nodesListPushFront(SNodeList* pList, SNodeptr pNode);
int32_t nodesListPushFront(SNodeList* pList, SNode* pNode);
SListCell* nodesListErase(SNodeList* pList, SListCell* pCell);
void nodesListInsertList(SNodeList* pTarget, SListCell* pPos, SNodeList* pSrc);
SNodeptr nodesListGetNode(SNodeList* pList, int32_t index);
SNode* nodesListGetNode(SNodeList* pList, int32_t index);
void nodesDestroyList(SNodeList* pList);
// Only clear the linked list structure, without releasing the elements inside
void nodesClearList(SNodeList* pList);
......@@ -275,9 +273,9 @@ void nodesClearList(SNodeList* pList);
typedef enum EDealRes { DEAL_RES_CONTINUE = 1, DEAL_RES_IGNORE_CHILD, DEAL_RES_ERROR, DEAL_RES_END } EDealRes;
typedef EDealRes (*FNodeWalker)(SNode* pNode, void* pContext);
void nodesWalkExpr(SNodeptr pNode, FNodeWalker walker, void* pContext);
void nodesWalkExpr(SNode* pNode, FNodeWalker walker, void* pContext);
void nodesWalkExprs(SNodeList* pList, FNodeWalker walker, void* pContext);
void nodesWalkExprPostOrder(SNodeptr pNode, FNodeWalker walker, void* pContext);
void nodesWalkExprPostOrder(SNode* pNode, FNodeWalker walker, void* pContext);
void nodesWalkExprsPostOrder(SNodeList* pList, FNodeWalker walker, void* pContext);
typedef EDealRes (*FNodeRewriter)(SNode** pNode, void* pContext);
......@@ -286,13 +284,13 @@ void nodesRewriteExprs(SNodeList* pList, FNodeRewriter rewriter, void* pContext)
void nodesRewriteExprPostOrder(SNode** pNode, FNodeRewriter rewriter, void* pContext);
void nodesRewriteExprsPostOrder(SNodeList* pList, FNodeRewriter rewriter, void* pContext);
bool nodesEqualNode(const SNodeptr a, const SNodeptr b);
bool nodesEqualNode(const SNode* a, const SNode* b);
SNodeptr nodesCloneNode(const SNodeptr pNode);
SNode* nodesCloneNode(const SNode* pNode);
SNodeList* nodesCloneList(const SNodeList* pList);
const char* nodesNodeName(ENodeType type);
int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_t* pLen);
int32_t nodesNodeToString(const SNode* pNode, bool format, char** pStr, int32_t* pLen);
int32_t nodesStringToNode(const char* pStr, SNode** pNode);
int32_t nodesListToString(const SNodeList* pList, bool format, char** pStr, int32_t* pLen);
......
include/libs/nodes/querynodes.h
浏览文件 @ c67e5722
......@@ -351,7 +351,6 @@ typedef struct SQuery {
int32_t placeholderNum;
SArray* pPlaceholderValues;
SNode* pPrepareRoot;
struct SParseMetaCache* pMetaCache;
} SQuery;
void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);
......
source/client/src/clientHb.c
浏览文件 @ c67e5722
......@@ -698,7 +698,7 @@ SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) {
return NULL;
}
taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
// taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
taosThreadMutexLock(&clientHbMgr.lock);
taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);
......
source/client/src/clientImpl.c
浏览文件 @ c67e5722
......@@ -233,9 +233,7 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
return TSDB_CODE_SUCCESS;
}
static SAppInstInfo* getAppInfo(SRequestObj* pRequest) {
return pRequest->pTscObj->pAppInfo;
}
static SAppInstInfo* getAppInfo(SRequestObj* pRequest) { return pRequest->pTscObj->pAppInfo; }
void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
SRetrieveTableRsp* pRsp = NULL;
......@@ -259,7 +257,7 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
}
pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
// pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
// pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
}
int32_t asyncExecDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
......@@ -404,16 +402,14 @@ int32_t scheduleAsyncQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNod
SQueryResult res = {.code = 0, .numOfRows = 0};
SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
.requestId = pRequest->requestId,
.requestObjRefId = pRequest->self
};
.requestObjRefId = pRequest->self};
SSchedulerReq req = {.pConn = &conn,
.pNodeList = pNodeList,
.pDag = pDag,
.sql = pRequest->sqlstr,
.startTs = pRequest->metric.start,
.fp = schdExecCallback,
.cbParam = &res
};
.cbParam = &res};
int32_t code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
......@@ -458,18 +454,16 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
SQueryResult res = {0};
SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
.requestId = pRequest->requestId,
.requestObjRefId = pRequest->self
};
.requestObjRefId = pRequest->self};
SSchedulerReq req = {.pConn = &conn,
.pNodeList = pNodeList,
.pDag = pDag,
.sql = pRequest->sqlstr,
.startTs = pRequest->metric.start,
.fp = NULL,
.cbParam = NULL
};
.cbParam = NULL};
int32_t code = schedulerExecJob(&req,&pRequest->body.queryJob, &res);
int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob, &res);
pRequest->body.resInfo.execRes = res.res;
if (code != TSDB_CODE_SUCCESS) {
......@@ -482,7 +476,8 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
return pRequest->code;
}
if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type || TDMT_VND_CREATE_TABLE == pRequest->type) {
if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type ||
TDMT_VND_CREATE_TABLE == pRequest->type) {
pRequest->body.resInfo.numOfRows = res.numOfRows;
if (pRequest->body.queryJob != 0) {
......@@ -495,7 +490,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
return pRequest->code;
}
int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet *epset) {
int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
int32_t code = 0;
SArray* pArray = NULL;
SSubmitRsp* pRsp = (SSubmitRsp*)res;
......@@ -532,7 +527,7 @@ _return:
return code;
}
int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet *epset) {
int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
int32_t code = 0;
SArray* pArray = NULL;
SArray* pTbArray = (SArray*)res;
......@@ -601,7 +596,7 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
break;
}
default:
tscError("0x%"PRIx64", invalid exec result for request type %d, reqId:0x%"PRIx64, pRequest->self,
tscError("0x%" PRIx64 ", invalid exec result for request type %d, reqId:0x%" PRIx64, pRequest->self,
pRequest->type, pRequest->requestId);
code = TSDB_CODE_APP_ERROR;
}
......@@ -610,13 +605,13 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
}
void schedulerExecCb(SQueryResult* pResult, void* param, int32_t code) {
SRequestObj* pRequest = (SRequestObj*) param;
SRequestObj* pRequest = (SRequestObj*)param;
pRequest->code = code;
STscObj* pTscObj = pRequest->pTscObj;
if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code)) {
tscDebug("0x%"PRIx64" client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%"PRIx64, pRequest->self, code, tstrerror(code),
pRequest->retry, pRequest->requestId);
tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%" PRIx64,
pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
pRequest->prevCode = code;
doAsyncQuery(pRequest, true);
return;
......@@ -727,30 +722,29 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery) {
if (TSDB_CODE_SUCCESS == code) {
code = qCreateQueryPlan(&cxt, &pRequest->body.pDag, pNodeList);
if (code) {
tscError("0x%"PRIx64" failed to create query plan, code:%s 0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId);
tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
pRequest->requestId);
}
}
if (TSDB_CODE_SUCCESS == code) {
SRequestConnInfo conn = {.pTrans = pAppInfo->pTransporter,
.requestId = pRequest->requestId,
.requestObjRefId = pRequest->self
};
SRequestConnInfo conn = {
.pTrans = pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
SSchedulerReq req = {.pConn = &conn,
.pNodeList = pNodeList,
.pDag = pRequest->body.pDag,
.sql = pRequest->sqlstr,
.startTs = pRequest->metric.start,
.fp = schedulerExecCb,
.cbParam = pRequest
};
.cbParam = pRequest};
code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
} else {
tscError("0x%"PRIx64" failed to create query plan, code:%s 0x%"PRIx64, pRequest->self, tstrerror(code), pRequest->requestId);
tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
pRequest->requestId);
pRequest->body.queryFp(pRequest->body.param, pRequest, code);
}
//todo not to be released here
// todo not to be released here
taosArrayDestroy(pNodeList);
break;
}
......@@ -997,9 +991,8 @@ void updateTargetEpSet(SMsgSendInfo* pSendInfo, STscObj* pTscObj, SRpcMsg* pMsg,
SEpSet* pOrig = &pTscObj->pAppInfo->mgmtEp.epSet;
SEp* pOrigEp = &pOrig->eps[pOrig->inUse];
SEp* pNewEp = &pEpSet->eps[pEpSet->inUse];
tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in client",
pOrig->inUse, pOrig->numOfEps, pOrigEp->fqdn, pOrigEp->port,
pEpSet->inUse, pEpSet->numOfEps, pNewEp->fqdn, pNewEp->port);
tscDebug("mnode epset updated from %d/%d=>%s:%d to %d/%d=>%s:%d in client", pOrig->inUse, pOrig->numOfEps,
pOrigEp->fqdn, pOrigEp->port, pEpSet->inUse, pEpSet->numOfEps, pNewEp->fqdn, pNewEp->port);
updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, pEpSet);
break;
case TARGET_TYPE_VNODE: {
......@@ -1415,14 +1408,14 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
p += sizeof(uint64_t);
// check fields
for(int32_t i = 0; i < numOfCols; ++i) {
int16_t type = *(int16_t*) p;
for (int32_t i = 0; i < numOfCols; ++i) {
int16_t type = *(int16_t*)p;
p += sizeof(int16_t);
int32_t bytes = *(int32_t*) p;
int32_t bytes = *(int32_t*)p;
p += sizeof(int32_t);
// ASSERT(type == pFields[i].type && bytes == pFields[i].bytes);
// ASSERT(type == pFields[i].type && bytes == pFields[i].bytes);
}
int32_t* colLength = (int32_t*)p;
......
source/client/src/clientMain.c
浏览文件 @ c67e5722
......@@ -30,7 +30,7 @@
#define TSC_VAR_RELEASED 0
static int32_t sentinel = TSC_VAR_NOT_RELEASE;
static int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt);
static int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt);
int taos_options(TSDB_OPTION option, const void *arg, ...) {
static int32_t lock = 0;
......@@ -177,8 +177,8 @@ TAOS_FIELD *taos_fetch_fields(TAOS_RES *res) {
return pResInfo->userFields;
}
static void syncQueryFn(void* param, void* res, int32_t code) {
SSyncQueryParam* pParam = param;
static void syncQueryFn(void *param, void *res, int32_t code) {
SSyncQueryParam *pParam = param;
pParam->pRequest = res;
pParam->pRequest->code = code;
......@@ -190,10 +190,10 @@ TAOS_RES *taos_query(TAOS *taos, const char *sql) {
return NULL;
}
STscObj* pTscObj = (STscObj*)taos;
STscObj *pTscObj = (STscObj *)taos;
#if SYNC_ON_TOP_OF_ASYNC
SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
SSyncQueryParam *param = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
tsem_init(&param->sem, 0, 0);
taos_query_a(pTscObj, sql, syncQueryFn, param);
......@@ -606,16 +606,30 @@ const char *taos_get_server_info(TAOS *taos) {
}
typedef struct SqlParseWrapper {
SParseContext* pCtx;
SParseContext *pCtx;
SCatalogReq catalogReq;
SRequestObj* pRequest;
SQuery* pQuery;
SRequestObj *pRequest;
SQuery *pQuery;
} SqlParseWrapper;
void retrieveMetaCallback(SMetaData* pResultMeta, void* param, int32_t code) {
SqlParseWrapper *pWrapper = (SqlParseWrapper*) param;
SQuery* pQuery = pWrapper->pQuery;
SRequestObj* pRequest = pWrapper->pRequest;
static void destorySqlParseWrapper(SqlParseWrapper *pWrapper) {
taosArrayDestroy(pWrapper->catalogReq.pDbVgroup);
taosArrayDestroy(pWrapper->catalogReq.pDbCfg);
taosArrayDestroy(pWrapper->catalogReq.pDbInfo);
taosArrayDestroy(pWrapper->catalogReq.pTableMeta);
taosArrayDestroy(pWrapper->catalogReq.pTableHash);
taosArrayDestroy(pWrapper->catalogReq.pUdf);
taosArrayDestroy(pWrapper->catalogReq.pIndex);
taosArrayDestroy(pWrapper->catalogReq.pUser);
taosArrayDestroy(pWrapper->catalogReq.pTableIndex);
taosMemoryFree(pWrapper->pCtx);
taosMemoryFree(pWrapper);
}
void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
SqlParseWrapper *pWrapper = (SqlParseWrapper *)param;
SQuery *pQuery = pWrapper->pQuery;
SRequestObj *pRequest = pWrapper->pRequest;
if (code == TSDB_CODE_SUCCESS) {
code = qAnalyseSqlSemantic(pWrapper->pCtx, &pWrapper->catalogReq, pResultMeta, pQuery);
......@@ -630,22 +644,22 @@ void retrieveMetaCallback(SMetaData* pResultMeta, void* param, int32_t code) {
TSWAP(pRequest->dbList, (pQuery)->pDbList);
TSWAP(pRequest->tableList, (pQuery)->pTableList);
taosMemoryFree(pWrapper);
destorySqlParseWrapper(pWrapper);
launchAsyncQuery(pRequest, pQuery);
} else {
destorySqlParseWrapper(pWrapper);
tscDebug("error happens, code:%d", code);
if (NEED_CLIENT_HANDLE_ERROR(code)) {
tscDebug("0x%"PRIx64" client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%"PRIx64, pRequest->self, code, tstrerror(code),
pRequest->retry, pRequest->requestId);
tscDebug("0x%" PRIx64 " client retry to handle the error, code:%d - %s, tryCount:%d, reqId:0x%" PRIx64,
pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId);
pRequest->prevCode = code;
doAsyncQuery(pRequest, true);
return;
}
// return to app directly
taosMemoryFree(pWrapper);
tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:0x%" PRIx64, pRequest->self, tstrerror(code),
pRequest->requestId);
tscError("0x%" PRIx64 " error occurs, code:%s, return to user app, reqId:0x%" PRIx64, pRequest->self,
tstrerror(code), pRequest->requestId);
pRequest->code = code;
pRequest->body.queryFp(pRequest->body.param, pRequest, code);
}
......@@ -682,7 +696,7 @@ void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param
doAsyncQuery(pRequest, false);
}
int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt) {
int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
const STscObj *pTscObj = pRequest->pTscObj;
*pCxt = taosMemoryCalloc(1, sizeof(SParseContext));
......@@ -690,7 +704,8 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt) {
return TSDB_CODE_OUT_OF_MEMORY;
}
**pCxt = (SParseContext){.requestId = pRequest->requestId,
**pCxt = (SParseContext){
.requestId = pRequest->requestId,
.requestRid = pRequest->self,
.acctId = pTscObj->acctId,
.db = pRequest->pDb,
......@@ -704,12 +719,13 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext** pCxt) {
.pUser = pTscObj->user,
.schemalessType = pTscObj->schemalessType,
.isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
.async = true,};
.async = true,
};
return TSDB_CODE_SUCCESS;
}
void doAsyncQuery(SRequestObj* pRequest, bool updateMetaForce) {
SParseContext* pCxt = NULL;
void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
SParseContext *pCxt = NULL;
STscObj *pTscObj = pRequest->pTscObj;
int32_t code = 0;
......@@ -753,23 +769,24 @@ void doAsyncQuery(SRequestObj* pRequest, bool updateMetaForce) {
.requestObjRefId = pCxt->requestRid,
.mgmtEps = pCxt->mgmtEpSet};
code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, pRequest->requestId,
&catalogReq, retrieveMetaCallback, pWrapper, &pRequest->body.queryJob);
code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, pRequest->requestId, &catalogReq, retrieveMetaCallback, pWrapper,
&pRequest->body.queryJob);
if (code == TSDB_CODE_SUCCESS) {
return;
}
_error:
tscError("0x%"PRIx64" error happens, code:%d - %s, reqId:0x%"PRIx64, pRequest->self, code, tstrerror(code), pRequest->requestId);
_error:
tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code),
pRequest->requestId);
terrno = code;
pRequest->code = code;
pRequest->body.queryFp(pRequest->body.param, pRequest, code);
}
static void fetchCallback(void* pResult, void* param, int32_t code) {
SRequestObj* pRequest = (SRequestObj*) param;
static void fetchCallback(void *pResult, void *param, int32_t code) {
SRequestObj *pRequest = (SRequestObj *)param;
SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
pResultInfo->pData = pResult;
pResultInfo->numOfRows = 0;
......@@ -785,7 +802,8 @@ static void fetchCallback(void* pResult, void* param, int32_t code) {
return;
}
pRequest->code = setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp*)pResultInfo->pData, pResultInfo->convertUcs4, false);
pRequest->code =
setQueryResultFromRsp(pResultInfo, (SRetrieveTableRsp *)pResultInfo->pData, pResultInfo->convertUcs4, false);
if (pRequest->code != TSDB_CODE_SUCCESS) {
pResultInfo->numOfRows = 0;
pRequest->code = code;
......@@ -801,7 +819,7 @@ static void fetchCallback(void* pResult, void* param, int32_t code) {
}
void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
ASSERT (res != NULL && fp != NULL);
ASSERT(res != NULL && fp != NULL);
SRequestObj *pRequest = res;
pRequest->body.fetchFp = fp;
......@@ -825,7 +843,7 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
schedulerAsyncFetchRows(pRequest->body.queryJob, fetchCallback, pRequest);
}
void taos_fetch_raw_block_a(TAOS_RES* res, __taos_async_fn_t fp, void* param) {
void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
ASSERT(res != NULL && fp != NULL);
SRequestObj *pRequest = res;
......@@ -838,9 +856,9 @@ void taos_fetch_raw_block_a(TAOS_RES* res, __taos_async_fn_t fp, void* param) {
taos_fetch_rows_a(res, fp, param);
}
const void* taos_get_raw_block(TAOS_RES* res) {
const void *taos_get_raw_block(TAOS_RES *res) {
ASSERT(res != NULL);
SRequestObj* pRequest = res;
SRequestObj *pRequest = res;
return pRequest->body.resInfo.pData;
}
......@@ -924,10 +942,9 @@ int taos_stmt_set_tags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
return stmtSetTbTags(stmt, tags);
}
int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name) { return taos_stmt_set_tbname(stmt, name); }
int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields) {
int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
if (stmt == NULL || NULL == fieldNum) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
......@@ -937,7 +954,7 @@ int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fiel
return stmtGetTagFields(stmt, fieldNum, fields);
}
int taos_stmt_get_col_fields(TAOS_STMT *stmt, int* fieldNum, TAOS_FIELD_E** fields) {
int taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields) {
if (stmt == NULL || NULL == fieldNum) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
......@@ -1102,4 +1119,3 @@ int taos_stmt_close(TAOS_STMT *stmt) {
return stmtClose(stmt);
}
source/client/src/clientSml.c
浏览文件 @ c67e5722
此差异已折叠。
source/common/src/tmsg.c
浏览文件 @ c67e5722
......@@ -692,6 +692,7 @@ int32_t tDeserializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq
void tFreeSMAltertbReq(SMAlterStbReq *pReq) {
taosArrayDestroy(pReq->pFields);
pReq->pFields = NULL;
taosMemoryFreeClear(pReq->comment);
}
int32_t tSerializeSEpSet(void *buf, int32_t bufLen, const SEpSet *pEpset) {
......@@ -4064,6 +4065,7 @@ int32_t tSerializeSCMCreateStreamReq(void *buf, int32_t bufLen, const SCMCreateS
if (tEncodeI32(&encoder, sqlLen) < 0) return -1;
if (tEncodeI32(&encoder, astLen) < 0) return -1;
if (tEncodeI8(&encoder, pReq->triggerType) < 0) return -1;
if (tEncodeI64(&encoder, pReq->maxDelay) < 0) return -1;
if (tEncodeI64(&encoder, pReq->watermark) < 0) return -1;
if (sqlLen > 0 && tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
if (astLen > 0 && tEncodeCStr(&encoder, pReq->ast) < 0) return -1;
......@@ -4090,6 +4092,7 @@ int32_t tDeserializeSCMCreateStreamReq(void *buf, int32_t bufLen, SCMCreateStrea
if (tDecodeI32(&decoder, &sqlLen) < 0) return -1;
if (tDecodeI32(&decoder, &astLen) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->triggerType) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->maxDelay) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->watermark) < 0) return -1;
if (sqlLen > 0) {
......
source/dnode/mnode/impl/src/mndScheduler.c
浏览文件 @ c67e5722
......@@ -77,7 +77,7 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
terrno = TSDB_CODE_QRY_INVALID_INPUT;
goto END;
}
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
int32_t opNum = LIST_LENGTH(inner->pNodeList);
if (opNum != 1) {
......@@ -85,7 +85,7 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
goto END;
}
SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
if (qSubPlanToString(plan, pDst, pDstLen) < 0) {
terrno = TSDB_CODE_QRY_INVALID_INPUT;
goto END;
......@@ -93,7 +93,7 @@ int32_t mndConvertRsmaTask(char** pDst, int32_t* pDstLen, const char* ast, int64
END:
if (pAst) nodesDestroyNode(pAst);
if (pPlan) nodesDestroyNode(pPlan);
if (pPlan) nodesDestroyNode((SNode*)pPlan);
return terrno;
}
......@@ -378,8 +378,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
SArray* taskInnerLevel = taosArrayInit(0, sizeof(void*));
taosArrayPush(pStream->tasks, &taskInnerLevel);
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
ASSERT(plan->subplanType == SUBPLAN_TYPE_MERGE);
pFinalTask = tNewSStreamTask(pStream->uid);
......@@ -407,8 +407,8 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
SArray* taskSourceLevel = taosArrayInit(0, sizeof(void*));
taosArrayPush(pStream->tasks, &taskSourceLevel);
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 1);
SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 1);
SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
void* pIter = NULL;
......@@ -449,9 +449,9 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
SArray* taskOneLevel = taosArrayInit(0, sizeof(void*));
taosArrayPush(pStream->tasks, &taskOneLevel);
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
ASSERT(LIST_LENGTH(inner->pNodeList) == 1);
SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
SSubplan* plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
ASSERT(plan->subplanType == SUBPLAN_TYPE_SCAN);
void* pIter = NULL;
......@@ -509,7 +509,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
return -1;
}
SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
SNodeListNode* inner = (SNodeListNode*)nodesListGetNode(pPlan->pSubplans, 0);
int32_t opNum = LIST_LENGTH(inner->pNodeList);
if (opNum != 1) {
......@@ -517,7 +517,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
terrno = TSDB_CODE_MND_INVALID_TOPIC_QUERY;
return -1;
}
plan = nodesListGetNode(inner->pNodeList, 0);
plan = (SSubplan*)nodesListGetNode(inner->pNodeList, 0);
}
ASSERT(pSub->unassignedVgs);
......
source/dnode/mnode/impl/src/mndStream.c
浏览文件 @ c67e5722
......@@ -235,10 +235,10 @@ static int32_t mndStreamGetPlanString(const char *ast, int8_t triggerType, int64
}
if (TSDB_CODE_SUCCESS == code) {
code = nodesNodeToString(pPlan, false, pStr, NULL);
code = nodesNodeToString((SNode*)pPlan, false, pStr, NULL);
}
nodesDestroyNode(pAst);
nodesDestroyNode(pPlan);
nodesDestroyNode((SNode*)pPlan);
terrno = code;
return code;
}
......
source/dnode/mnode/impl/src/mndTopic.c
浏览文件 @ c67e5722
......@@ -387,7 +387,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
return -1;
}
if (nodesNodeToString(pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
if (nodesNodeToString((SNode*)pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
taosMemoryFree(topicObj.ast);
taosMemoryFree(topicObj.sql);
......
source/libs/command/src/explain.c
浏览文件 @ c67e5722
......@@ -211,7 +211,7 @@ int32_t qExplainGenerateResChildren(SPhysiNode *pNode, SExplainGroup *group, SNo
SExplainResNode *pResNode = NULL;
FOREACH(node, pPhysiChildren) {
QRY_ERR_RET(qExplainGenerateResNode((SPhysiNode *)node, group, &pResNode));
QRY_ERR_RET(nodesListAppend(*pChildren, pResNode));
QRY_ERR_RET(nodesListAppend(*pChildren, (SNode*)pResNode));
}
return TSDB_CODE_SUCCESS;
......@@ -715,7 +715,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_NEW(level + 1, "Sort Key: ");
if (pResNode->pExecInfo) {
for (int32_t i = 0; i < LIST_LENGTH(pSortNode->pSortKeys); ++i) {
SOrderByExprNode *ptn = nodesListGetNode(pSortNode->pSortKeys, i);
SOrderByExprNode *ptn = (SOrderByExprNode*)nodesListGetNode(pSortNode->pSortKeys, i);
EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr));
}
}
......@@ -1039,7 +1039,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_NEW(level + 1, "Merge Key: ");
if (pResNode->pExecInfo) {
for (int32_t i = 0; i < LIST_LENGTH(pMergeNode->pMergeKeys); ++i) {
SOrderByExprNode *ptn = nodesListGetNode(pMergeNode->pMergeKeys, i);
SOrderByExprNode *ptn = (SOrderByExprNode*)nodesListGetNode(pMergeNode->pMergeKeys, i);
EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr));
}
}
......@@ -1078,7 +1078,7 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
EXPLAIN_ROW_NEW(level + 1, EXPLAIN_MERGE_KEYS_FORMAT);
for (int32_t i = 0; i < LIST_LENGTH(pMergeNode->pMergeKeys); ++i) {
SOrderByExprNode *ptn = nodesListGetNode(pMergeNode->pMergeKeys, i);
SOrderByExprNode *ptn = (SOrderByExprNode*)nodesListGetNode(pMergeNode->pMergeKeys, i);
EXPLAIN_ROW_APPEND("%s ", nodesGetNameFromColumnNode(ptn->pExpr));
}
EXPLAIN_ROW_END();
......
source/libs/executor/src/executorimpl.c
浏览文件 @ c67e5722
......@@ -202,7 +202,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {{0}};
SSlotDescNode* pDescNode = nodesListGetNode(pNode->pSlots, i);
SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
// if (!pDescNode->output) { // todo disable it temporarily
// continue;
// }
......@@ -2920,7 +2920,7 @@ static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo*
}
for (int32_t i = 0; i < numOfSources; ++i) {
SNodeListNode* pNode = nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
SNodeListNode* pNode = (SNodeListNode*)nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
taosArrayPush(pInfo->pSources, pNode);
}
......@@ -4494,7 +4494,7 @@ SExprInfo* createExprInfo(SNodeList* pNodeList, SNodeList* pGroupKeys, int32_t*
if (NULL == res) { // todo handle error
} else {
res->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
nodesListAppend(pFuncNode->pParameterList, res);
nodesListAppend(pFuncNode->pParameterList, (SNode*)res);
}
}
#endif
......
source/libs/executor/src/timewindowoperator.c
浏览文件 @ c67e5722
......@@ -1378,7 +1378,7 @@ void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
taosMemoryFreeClear(pChildOp);
}
}
nodesDestroyNode(pInfo->pPhyNode);
nodesDestroyNode((SNode*)pInfo->pPhyNode);
}
static bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
......@@ -2157,7 +2157,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
pInfo->pUpdateRes->info.type = STREAM_REPROCESS;
blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
pInfo->pPhyNode = nodesCloneNode(pPhyNode);
pInfo->pPhyNode = (SPhysiNode*)nodesCloneNode((SNode*)pPhyNode);
pOperator->name = "StreamFinalIntervalOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL;
......
source/libs/function/inc/functionMgtInt.h
浏览文件 @ c67e5722
......@@ -42,6 +42,7 @@ extern "C" {
#define FUNC_MGT_SELECT_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(13)
#define FUNC_MGT_REPEAT_SCAN_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(14)
#define FUNC_MGT_FORBID_FILL_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(15)
#define FUNC_MGT_FORBID_STREAM_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(16)
#define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)
......
source/libs/function/src/builtins.c
浏览文件 @ c67e5722
......@@ -1274,7 +1274,7 @@ void static addTimezoneParam(SNodeList* pList) {
varDataSetLen(pVal->datum.p, len);
strncpy(varDataVal(pVal->datum.p), pVal->literal, len);
nodesListAppend(pList, pVal);
nodesListAppend(pList, (SNode*)pVal);
}
static int32_t translateToIso8601(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
......@@ -1363,7 +1363,7 @@ static int32_t translateToJson(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
SExprNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
SExprNode* pPara = (SExprNode*)nodesListGetNode(pFunc->pParameterList, 0);
if (QUERY_NODE_VALUE != nodeType(pPara) || (!IS_VAR_DATA_TYPE(pPara->resType.type))) {
return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
}
......
source/libs/function/src/functionMgt.c
浏览文件 @ c67e5722
......@@ -161,6 +161,8 @@ bool fmIsUserDefinedFunc(int32_t funcId) { return funcId > FUNC_UDF_ID_START; }
bool fmIsForbidFillFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_FILL_FUNC); }
bool fmIsForbidStreamFunc(int32_t funcId) { return isSpecificClassifyFunc(funcId, FUNC_MGT_FORBID_STREAM_FUNC); }
void fmFuncMgtDestroy() {
void* m = gFunMgtService.pFuncNameHashTable;
if (m != NULL && atomic_val_compare_exchange_ptr((void**)&gFunMgtService.pFuncNameHashTable, m, 0) == m) {
......@@ -239,7 +241,7 @@ static int32_t getFuncInfo(SFunctionNode* pFunc) {
}
static SFunctionNode* createFunction(const char* pName, SNodeList* pParameterList) {
SFunctionNode* pFunc = nodesMakeNode(QUERY_NODE_FUNCTION);
SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
if (NULL == pFunc) {
return NULL;
}
......@@ -247,14 +249,14 @@ static SFunctionNode* createFunction(const char* pName, SNodeList* pParameterLis
pFunc->pParameterList = pParameterList;
if (TSDB_CODE_SUCCESS != getFuncInfo(pFunc)) {
pFunc->pParameterList = NULL;
nodesDestroyNode(pFunc);
nodesDestroyNode((SNode*)pFunc);
return NULL;
}
return pFunc;
}
static SColumnNode* createColumnByFunc(const SFunctionNode* pFunc) {
SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
return NULL;
}
......@@ -291,7 +293,7 @@ static int32_t createPartialFunction(const SFunctionNode* pSrcFunc, SFunctionNod
static int32_t createMergeFunction(const SFunctionNode* pSrcFunc, const SFunctionNode* pPartialFunc,
SFunctionNode** pMergeFunc) {
SNodeList* pParameterList = NULL;
nodesListMakeStrictAppend(&pParameterList, createColumnByFunc(pPartialFunc));
nodesListMakeStrictAppend(&pParameterList, (SNode*)createColumnByFunc(pPartialFunc));
*pMergeFunc = createFunction(funcMgtBuiltins[pSrcFunc->funcId].pMergeFunc, pParameterList);
if (NULL == *pMergeFunc) {
nodesDestroyList(pParameterList);
......@@ -316,8 +318,8 @@ int32_t fmGetDistMethod(const SFunctionNode* pFunc, SFunctionNode** pPartialFunc
}
if (TSDB_CODE_SUCCESS != code) {
nodesDestroyNode(*pPartialFunc);
nodesDestroyNode(*pMergeFunc);
nodesDestroyNode((SNode*)*pPartialFunc);
nodesDestroyNode((SNode*)*pMergeFunc);
}
return code;
......
source/libs/nodes/src/nodesCloneFuncs.c
浏览文件 @ c67e5722
......@@ -40,6 +40,10 @@
break; \
} \
(pDst)->fldname = strdup((pSrc)->fldname); \
if (NULL == (pDst)->fldname) { \
nodesDestroyNode((SNode*)pDst); \
return NULL; \
} \
} while (0)
#define CLONE_NODE_FIELD(fldname) \
......@@ -49,7 +53,19 @@
} \
(pDst)->fldname = nodesCloneNode((pSrc)->fldname); \
if (NULL == (pDst)->fldname) { \
nodesDestroyNode((SNode*)(pDst)); \
nodesDestroyNode((SNode*)pDst); \
return NULL; \
} \
} while (0)
#define CLONE_NODE_FIELD_EX(fldname, nodePtrType) \
do { \
if (NULL == (pSrc)->fldname) { \
break; \
} \
(pDst)->fldname = (nodePtrType)nodesCloneNode((SNode*)(pSrc)->fldname); \
if (NULL == (pDst)->fldname) { \
nodesDestroyNode((SNode*)pDst); \
return NULL; \
} \
} while (0)
......@@ -61,7 +77,7 @@
} \
(pDst)->fldname = nodesCloneList((pSrc)->fldname); \
if (NULL == (pDst)->fldname) { \
nodesDestroyNode((SNode*)(pDst)); \
nodesDestroyNode((SNode*)pDst); \
return NULL; \
} \
} while (0)
......@@ -73,7 +89,7 @@
} \
(pDst)->fldname = cloneFunc((pSrc)->fldname); \
if (NULL == (pDst)->fldname) { \
nodesDestroyNode((SNode*)(pDst)); \
nodesDestroyNode((SNode*)pDst); \
return NULL; \
} \
} while (0)
......@@ -81,6 +97,7 @@
#define COPY_BASE_OBJECT_FIELD(fldname, copyFunc) \
do { \
if (NULL == copyFunc(&((pSrc)->fldname), &((pDst)->fldname))) { \
nodesDestroyNode((SNode*)pDst); \
return NULL; \
} \
} while (0)
......@@ -147,7 +164,7 @@ static SNode* valueNodeCopy(const SValueNode* pSrc, SValueNode* pDst) {
int32_t len = varDataTLen(pSrc->datum.p) + 1;
pDst->datum.p = taosMemoryCalloc(1, len);
if (NULL == pDst->datum.p) {
nodesDestroyNode(pDst);
nodesDestroyNode((SNode*)pDst);
return NULL;
}
memcpy(pDst->datum.p, pSrc->datum.p, len);
......@@ -275,8 +292,8 @@ static SNode* stateWindowNodeCopy(const SStateWindowNode* pSrc, SStateWindowNode
}
static SNode* sessionWindowNodeCopy(const SSessionWindowNode* pSrc, SSessionWindowNode* pDst) {
CLONE_NODE_FIELD(pCol);
CLONE_NODE_FIELD(pGap);
CLONE_NODE_FIELD_EX(pCol, SColumnNode*);
CLONE_NODE_FIELD_EX(pGap, SValueNode*);
return (SNode*)pDst;
}
......@@ -442,7 +459,7 @@ static SNode* logicIndefRowsFuncCopy(const SIndefRowsFuncLogicNode* pSrc, SIndef
static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
COPY_OBJECT_FIELD(id, sizeof(SSubplanId));
CLONE_NODE_FIELD(pNode);
CLONE_NODE_FIELD_EX(pNode, SLogicNode*);
COPY_SCALAR_FIELD(subplanType);
COPY_SCALAR_FIELD(level);
COPY_SCALAR_FIELD(splitFlag);
......@@ -450,7 +467,7 @@ static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
}
static SNode* physiNodeCopy(const SPhysiNode* pSrc, SPhysiNode* pDst) {
CLONE_NODE_FIELD(pOutputDataBlockDesc);
CLONE_NODE_FIELD_EX(pOutputDataBlockDesc, SDataBlockDescNode*);
CLONE_NODE_FIELD(pConditions);
CLONE_NODE_LIST_FIELD(pChildren);
return (SNode*)pDst;
......@@ -555,8 +572,8 @@ static SNode* selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
CLONE_NODE_LIST_FIELD(pGroupByList);
CLONE_NODE_FIELD(pHaving);
CLONE_NODE_LIST_FIELD(pOrderByList);
CLONE_NODE_FIELD(pLimit);
CLONE_NODE_FIELD(pLimit);
CLONE_NODE_FIELD_EX(pLimit, SLimitNode*);
CLONE_NODE_FIELD_EX(pLimit, SLimitNode*);
COPY_CHAR_ARRAY_FIELD(stmtName);
COPY_SCALAR_FIELD(precision);
COPY_SCALAR_FIELD(isEmptyResult);
......@@ -566,7 +583,7 @@ static SNode* selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
return (SNode*)pDst;
}
SNodeptr nodesCloneNode(const SNodeptr pNode) {
SNode* nodesCloneNode(const SNode* pNode) {
if (NULL == pNode) {
return NULL;
}
......
source/libs/nodes/src/nodesCodeFuncs.c
浏览文件 @ c67e5722
......@@ -4205,7 +4205,7 @@ static int32_t jsonToNodeObject(const SJson* pJson, const char* pName, SNode** p
return makeNodeByJson(pJsonNode, pNode);
}
int32_t nodesNodeToString(const SNodeptr pNode, bool format, char** pStr, int32_t* pLen) {
int32_t nodesNodeToString(const SNode* pNode, bool format, char** pStr, int32_t* pLen) {
if (NULL == pNode || NULL == pStr) {
terrno = TSDB_CODE_FAILED;
return TSDB_CODE_FAILED;
......
source/libs/nodes/src/nodesEqualFuncs.c
浏览文件 @ c67e5722
......@@ -137,7 +137,7 @@ static bool functionNodeEqual(const SFunctionNode* a, const SFunctionNode* b) {
return true;
}
bool nodesEqualNode(const SNodeptr a, const SNodeptr b) {
bool nodesEqualNode(const SNode* a, const SNode* b) {
if (a == b) {
return true;
}
......
source/libs/nodes/src/nodesTraverseFuncs.c
浏览文件 @ c67e5722
......@@ -168,7 +168,7 @@ static EDealRes walkExprs(SNodeList* pNodeList, ETraversalOrder order, FNodeWalk
return DEAL_RES_CONTINUE;
}
void nodesWalkExpr(SNodeptr pNode, FNodeWalker walker, void* pContext) {
void nodesWalkExpr(SNode* pNode, FNodeWalker walker, void* pContext) {
(void)walkExpr(pNode, TRAVERSAL_PREORDER, walker, pContext);
}
......@@ -176,7 +176,7 @@ void nodesWalkExprs(SNodeList* pNodeList, FNodeWalker walker, void* pContext) {
(void)walkExprs(pNodeList, TRAVERSAL_PREORDER, walker, pContext);
}
void nodesWalkExprPostOrder(SNodeptr pNode, FNodeWalker walker, void* pContext) {
void nodesWalkExprPostOrder(SNode* pNode, FNodeWalker walker, void* pContext) {
(void)walkExpr(pNode, TRAVERSAL_POSTORDER, walker, pContext);
}
......
source/libs/parser/inc/parInt.h
浏览文件 @ c67e5722
......@@ -27,12 +27,12 @@ extern "C" {
#define QUERY_SMA_OPTIMIZE_DISABLE 0
#define QUERY_SMA_OPTIMIZE_ENABLE 1
int32_t parseInsertSyntax(SParseContext* pContext, SQuery** pQuery);
int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery);
int32_t parseInsertSyntax(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache);
int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache);
int32_t parse(SParseContext* pParseCxt, SQuery** pQuery);
int32_t collectMetaKey(SParseContext* pParseCxt, SQuery* pQuery);
int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery);
int32_t translate(SParseContext* pParseCxt, SQuery* pQuery);
int32_t collectMetaKey(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache);
int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache);
int32_t translate(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache);
int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
int32_t calculateConstant(SParseContext* pParseCxt, SQuery* pQuery);
......
source/libs/parser/inc/parUtil.h
浏览文件 @ c67e5722
......@@ -87,6 +87,7 @@ int32_t getUserAuthFromCache(SParseMetaCache* pMetaCache, const char* pUser, con
bool* pPass);
int32_t getUdfInfoFromCache(SParseMetaCache* pMetaCache, const char* pFunc, SFuncInfo* pInfo);
int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName, SArray** pIndexes);
void destoryParseMetaCache(SParseMetaCache* pMetaCache);
#ifdef __cplusplus
}
......
source/libs/parser/inc/sql.y
浏览文件 @ c67e5722
......@@ -461,6 +461,7 @@ into_opt(A) ::= INTO full_table_name(B).
stream_options(A) ::= . { A = createStreamOptions(pCxt); }
stream_options(A) ::= stream_options(B) TRIGGER AT_ONCE. { ((SStreamOptions*)B)->triggerType = STREAM_TRIGGER_AT_ONCE; A = B; }
stream_options(A) ::= stream_options(B) TRIGGER WINDOW_CLOSE. { ((SStreamOptions*)B)->triggerType = STREAM_TRIGGER_WINDOW_CLOSE; A = B; }
stream_options(A) ::= stream_options(B) TRIGGER MAX_DELAY duration_literal(C). { ((SStreamOptions*)B)->triggerType = STREAM_TRIGGER_MAX_DELAY; ((SStreamOptions*)B)->pDelay = releaseRawExprNode(pCxt, C); A = B; }
stream_options(A) ::= stream_options(B) WATERMARK duration_literal(C). { ((SStreamOptions*)B)->pWatermark = releaseRawExprNode(pCxt, C); A = B; }
/************************************************ kill connection/query ***********************************************/
......
source/libs/parser/src/parAuthenticator.c
浏览文件 @ c67e5722
......@@ -105,7 +105,7 @@ static int32_t authQuery(SAuthCxt* pCxt, SNode* pStmt) {
return TSDB_CODE_SUCCESS;
}
int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery) {
SAuthCxt cxt = {.pParseCxt = pParseCxt, .pMetaCache = pQuery->pMetaCache, .errCode = TSDB_CODE_SUCCESS};
int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery, SParseMetaCache* pMetaCache) {
SAuthCxt cxt = {.pParseCxt = pParseCxt, .pMetaCache = pMetaCache, .errCode = TSDB_CODE_SUCCESS};
return authQuery(&cxt, pQuery->pRoot);
}
source/libs/parser/src/parInsert.c
浏览文件 @ c67e5722
此差异已折叠。
source/libs/parser/src/parUtil.c
浏览文件 @ c67e5722
此差异已折叠。
source/libs/parser/src/parser.c
浏览文件 @ c67e5722
此差异已折叠。
source/libs/parser/src/sql.c
浏览文件 @ c67e5722
此差异已折叠。
source/libs/planner/src/planUtil.c
浏览文件 @ c67e5722
此差异已折叠。
source/libs/planner/src/planner.c
浏览文件 @ c67e5722
此差异已折叠。
source/libs/qworker/src/qwUtil.c
浏览文件 @ c67e5722
此差异已折叠。
source/util/src/thash.c
浏览文件 @ c67e5722
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或