Skip to content

T
TDengine

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

通知 1184
Star 22015
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 54fe93f7 编写于 作者: D dapan1121
浏览文件
操作

stmt query

上级 5935c9cd
隐藏空白更改
内联 并排
Showing with 285 addition and 148 deletion
source/client/inc/clientStmt.h
浏览文件 @ 54fe93f7
...@@ -46,11 +46,12 @@ typedef struct SStmtTableCache { ...@@ -46,11 +46,12 @@ typedef struct SStmtTableCache {
void* boundTags; void* boundTags;
} SStmtTableCache; } SStmtTableCache;
typedef struct SQueryFields { typedef struct SStmtQueryResInfo {
TAOS_FIELD* fields; TAOS_FIELD* fields;
TAOS_FIELD* userFields; TAOS_FIELD* userFields;
uint32_t numOfCols; uint32_t numOfCols;
} SQueryFields; int32_t precision;
} SStmtQueryResInfo;
typedef struct SStmtBindInfo { typedef struct SStmtBindInfo {
bool needParse; bool needParse;
...@@ -72,17 +73,17 @@ typedef struct SStmtExecInfo { ...@@ -72,17 +73,17 @@ typedef struct SStmtExecInfo {
} SStmtExecInfo; } SStmtExecInfo;
typedef struct SStmtSQLInfo { typedef struct SStmtSQLInfo {
STMT_TYPE type; STMT_TYPE type;
STMT_STATUS status; STMT_STATUS status;
bool autoCreate; bool autoCreate;
uint64_t runTimes; uint64_t runTimes;
SHashObj* pTableCache; //SHash<SStmtTableCache> SHashObj* pTableCache; //SHash<SStmtTableCache>
SQuery* pQuery; SQuery* pQuery;
char* sqlStr; char* sqlStr;
int32_t sqlLen; int32_t sqlLen;
SArray* nodeList; SArray* nodeList;
SQueryPlan* pQueryPlan; SQueryPlan* pQueryPlan;
SQueryFields fields; SStmtQueryResInfo queryRes;
} SStmtSQLInfo; } SStmtSQLInfo;
typedef struct STscStmt { typedef struct STscStmt {
......
source/client/src/clientStmt.c
浏览文件 @ 54fe93f7
...@@ -74,17 +74,44 @@ int32_t stmtGetTbName(TAOS_STMT *stmt, char **tbName) { ...@@ -74,17 +74,44 @@ int32_t stmtGetTbName(TAOS_STMT *stmt, char **tbName) {
} }
int32_t stmtBackupQueryFields(STscStmt* pStmt) { int32_t stmtBackupQueryFields(STscStmt* pStmt) {
SQueryFields *pFields = &pStmt->sql.fields; SStmtQueryResInfo *pRes = &pStmt->sql.queryRes;
int32_t size = pFields->numOfCols * sizeof(TAOS_FIELD); pRes->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols;
pRes->precision = pStmt->exec.pRequest->body.resInfo.precision;
pFields->numOfCols = pStmt->exec.pRequest->body.resInfo.numOfCols; int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
pFields->fields = taosMemoryMalloc(size); pRes->fields = taosMemoryMalloc(size);
pFields->userFields = taosMemoryMalloc(size); pRes->userFields = taosMemoryMalloc(size);
if (NULL == pFields->fields || NULL == pFields->userFields) { if (NULL == pRes->fields || NULL == pRes->userFields) {
STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY); STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY);
} }
memcpy(pFields->fields, pStmt->exec.pRequest->body.resInfo.fields, size); memcpy(pRes->fields, pStmt->exec.pRequest->body.resInfo.fields, size);
memcpy(pFields->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size); memcpy(pRes->userFields, pStmt->exec.pRequest->body.resInfo.userFields, size);
return TSDB_CODE_SUCCESS;
}
int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
SStmtQueryResInfo *pRes = &pStmt->sql.queryRes;
int32_t size = pRes->numOfCols * sizeof(TAOS_FIELD);
pStmt->exec.pRequest->body.resInfo.numOfCols = pRes->numOfCols;
pStmt->exec.pRequest->body.resInfo.precision = pRes->precision;
if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
pStmt->exec.pRequest->body.resInfo.fields = taosMemoryMalloc(size);
if (NULL == pStmt->exec.pRequest->body.resInfo.fields) {
STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY);
}
memcpy(pStmt->exec.pRequest->body.resInfo.fields, pRes->fields, size);
}
if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
pStmt->exec.pRequest->body.resInfo.userFields = taosMemoryMalloc(size);
if (NULL == pStmt->exec.pRequest->body.resInfo.userFields) {
STMT_ERR_RET(TSDB_CODE_TSC_OUT_OF_MEMORY);
}
memcpy(pStmt->exec.pRequest->body.resInfo.userFields, pRes->userFields, size);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
...@@ -235,6 +262,8 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) { ...@@ -235,6 +262,8 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
} }
int32_t stmtCleanSQLInfo(STscStmt* pStmt) { int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
taosMemoryFree(pStmt->sql.queryRes.fields);
taosMemoryFree(pStmt->sql.queryRes.userFields);
taosMemoryFree(pStmt->sql.sqlStr); taosMemoryFree(pStmt->sql.sqlStr);
qDestroyQuery(pStmt->sql.pQuery); qDestroyQuery(pStmt->sql.pQuery);
qDestroyQueryPlan(pStmt->sql.pQueryPlan); qDestroyQueryPlan(pStmt->sql.pQueryPlan);
...@@ -497,6 +526,8 @@ int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) { ...@@ -497,6 +526,8 @@ int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) {
pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag; pStmt->sql.pQueryPlan = pStmt->exec.pRequest->body.pDag;
pStmt->exec.pRequest->body.pDag = NULL; pStmt->exec.pRequest->body.pDag = NULL;
STMT_ERR_RET(stmtBackupQueryFields(pStmt)); STMT_ERR_RET(stmtBackupQueryFields(pStmt));
} else {
STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
} }
STMT_RET(qStmtBindParam(pStmt->sql.pQueryPlan, bind, colIdx, pStmt->exec.pRequest->requestId)); STMT_RET(qStmtBindParam(pStmt->sql.pQueryPlan, bind, colIdx, pStmt->exec.pRequest->requestId));
......
source/dnode/mnode/impl/inc/mndInt.h
浏览文件 @ 54fe93f7
...@@ -47,7 +47,7 @@ typedef int32_t (*MndInitFp)(SMnode *pMnode); ...@@ -47,7 +47,7 @@ typedef int32_t (*MndInitFp)(SMnode *pMnode);
typedef void (*MndCleanupFp)(SMnode *pMnode); typedef void (*MndCleanupFp)(SMnode *pMnode);
typedef int32_t (*ShowRetrieveFp)(SNodeMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows); typedef int32_t (*ShowRetrieveFp)(SNodeMsg *pMsg, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
typedef void (*ShowFreeIterFp)(SMnode *pMnode, void *pIter); typedef void (*ShowFreeIterFp)(SMnode *pMnode, void *pIter);
typedef struct SQWorkerMgmt SQHandle; typedef struct SQWorker SQHandle;
typedef struct { typedef struct {
const char *name; const char *name;
......
source/dnode/qnode/inc/qndInt.h
浏览文件 @ 54fe93f7
...@@ -29,7 +29,7 @@ ...@@ -29,7 +29,7 @@
extern "C" { extern "C" {
#endif #endif
typedef struct SQWorkerMgmt SQHandle; typedef struct SQWorker SQHandle;
typedef struct SQnode { typedef struct SQnode {
int32_t qndId; int32_t qndId;
......
source/dnode/vnode/src/inc/vnodeInt.h
浏览文件 @ 54fe93f7
...@@ -52,7 +52,7 @@ typedef struct STsdb STsdb; ...@@ -52,7 +52,7 @@ typedef struct STsdb STsdb;
typedef struct STQ STQ; typedef struct STQ STQ;
typedef struct SVState SVState; typedef struct SVState SVState;
typedef struct SVBufPool SVBufPool; typedef struct SVBufPool SVBufPool;
typedef struct SQWorkerMgmt SQHandle; typedef struct SQWorker SQHandle;
#define VNODE_META_DIR "meta" #define VNODE_META_DIR "meta"
#define VNODE_TSDB_DIR "tsdb" #define VNODE_TSDB_DIR "tsdb"
......
source/libs/planner/src/planOptimizer.c
浏览文件 @ 54fe93f7
...@@ -206,6 +206,8 @@ static int32_t cpdMergeCond(SNode** pDst, SNode** pSrc) { ...@@ -206,6 +206,8 @@ static int32_t cpdMergeCond(SNode** pDst, SNode** pSrc) {
if (NULL == pLogicCond) { if (NULL == pLogicCond) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
pLogicCond->node.resType.type = TSDB_DATA_TYPE_BOOL;
pLogicCond->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_BOOL].bytes;
pLogicCond->condType = LOGIC_COND_TYPE_AND; pLogicCond->condType = LOGIC_COND_TYPE_AND;
int32_t code = nodesListMakeAppend(&pLogicCond->pParameterList, *pSrc); int32_t code = nodesListMakeAppend(&pLogicCond->pParameterList, *pSrc);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
......
source/libs/qworker/inc/qworkerInt.h
浏览文件 @ 54fe93f7
...@@ -23,6 +23,7 @@ extern "C" { ...@@ -23,6 +23,7 @@ extern "C" {
#include "qworker.h" #include "qworker.h"
#include "tlockfree.h" #include "tlockfree.h"
#include "ttimer.h" #include "ttimer.h"
#include "tref.h"
#define QW_DEFAULT_SCHEDULER_NUMBER 10000 #define QW_DEFAULT_SCHEDULER_NUMBER 10000
#define QW_DEFAULT_TASK_NUMBER 10000 #define QW_DEFAULT_TASK_NUMBER 10000
...@@ -85,6 +86,11 @@ typedef struct SQWMsg { ...@@ -85,6 +86,11 @@ typedef struct SQWMsg {
SQWConnInfo connInfo; SQWConnInfo connInfo;
} SQWMsg; } SQWMsg;
typedef struct SQWHbParam {
int32_t qwrId;
int64_t refId;
} SQWHbParam;
typedef struct SQWHbInfo { typedef struct SQWHbInfo {
SSchedulerHbRsp rsp; SSchedulerHbRsp rsp;
SQWConnInfo connInfo; SQWConnInfo connInfo;
...@@ -137,7 +143,8 @@ typedef struct SQWSchStatus { ...@@ -137,7 +143,8 @@ typedef struct SQWSchStatus {
} SQWSchStatus; } SQWSchStatus;
// Qnode/Vnode level task management // Qnode/Vnode level task management
typedef struct SQWorkerMgmt { typedef struct SQWorker {
int64_t refId;
SQWorkerCfg cfg; SQWorkerCfg cfg;
int8_t nodeType; int8_t nodeType;
int32_t nodeId; int32_t nodeId;
...@@ -148,9 +155,15 @@ typedef struct SQWorkerMgmt { ...@@ -148,9 +155,15 @@ typedef struct SQWorkerMgmt {
SHashObj *schHash; // key: schedulerId, value: SQWSchStatus SHashObj *schHash; // key: schedulerId, value: SQWSchStatus
SHashObj *ctxHash; // key: queryId+taskId, value: SQWTaskCtx SHashObj *ctxHash; // key: queryId+taskId, value: SQWTaskCtx
SMsgCb msgCb; SMsgCb msgCb;
} SQWorker;
typedef struct SQWorkerMgmt {
SRWLatch lock;
int32_t qwRef;
int32_t qwNum;
} SQWorkerMgmt; } SQWorkerMgmt;
#define QW_FPARAMS_DEF SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int64_t rId #define QW_FPARAMS_DEF SQWorker *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int64_t rId
#define QW_IDS() sId, qId, tId, rId #define QW_IDS() sId, qId, tId, rId
#define QW_FPARAMS() mgmt, QW_IDS() #define QW_FPARAMS() mgmt, QW_IDS()
...@@ -209,13 +222,13 @@ typedef struct SQWorkerMgmt { ...@@ -209,13 +222,13 @@ typedef struct SQWorkerMgmt {
} \ } \
} while (0) } while (0)
#define QW_ELOG(param, ...) qError("QW:%p " param, mgmt, __VA_ARGS__) #define QW_ELOG(_param, ...) qError("QW:%p " _param, mgmt, __VA_ARGS__)
#define QW_DLOG(param, ...) qDebug("QW:%p " param, mgmt, __VA_ARGS__) #define QW_DLOG(_param, ...) qDebug("QW:%p " _param, mgmt, __VA_ARGS__)
#define QW_DUMP(param, ...) \ #define QW_DUMP(_param, ...) \
do { \ do { \
if (gQWDebug.dumpEnable) { \ if (gQWDebug.dumpEnable) { \
qDebug("QW:%p " param, mgmt, __VA_ARGS__); \ qDebug("QW:%p " _param, mgmt, __VA_ARGS__); \
} \ } \
} while (0) } while (0)
...@@ -282,6 +295,14 @@ typedef struct SQWorkerMgmt { ...@@ -282,6 +295,14 @@ typedef struct SQWorkerMgmt {
} \ } \
} while (0) } while (0)
extern SQWorkerMgmt gQwMgmt;
FORCE_INLINE SQWorker *qwAcquire(int64_t refId) { return (SQWorker *)taosAcquireRef(atomic_load_32(&gQwMgmt.qwRef), refId); }
FORCE_INLINE int32_t qwRelease(int64_t refId) { return taosReleaseRef(gQwMgmt.qwRef, refId); }
#ifdef __cplusplus #ifdef __cplusplus
} }
#endif #endif
......
source/libs/qworker/inc/qworkerMsg.h
浏览文件 @ 54fe93f7
...@@ -28,7 +28,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg); ...@@ -28,7 +28,7 @@ int32_t qwProcessCQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg);
int32_t qwProcessReady(QW_FPARAMS_DEF, SQWMsg *qwMsg); int32_t qwProcessReady(QW_FPARAMS_DEF, SQWMsg *qwMsg);
int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg); int32_t qwProcessFetch(QW_FPARAMS_DEF, SQWMsg *qwMsg);
int32_t qwProcessDrop(QW_FPARAMS_DEF, SQWMsg *qwMsg); int32_t qwProcessDrop(QW_FPARAMS_DEF, SQWMsg *qwMsg);
int32_t qwProcessHb(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req); int32_t qwProcessHb(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req);
int32_t qwBuildAndSendDropRsp(SQWConnInfo *pConn, int32_t code); int32_t qwBuildAndSendDropRsp(SQWConnInfo *pConn, int32_t code);
int32_t qwBuildAndSendCancelRsp(SQWConnInfo *pConn, int32_t code); int32_t qwBuildAndSendCancelRsp(SQWConnInfo *pConn, int32_t code);
...@@ -41,10 +41,10 @@ int32_t qwBuildAndSendQueryRsp(SQWConnInfo *pConn, int32_t code); ...@@ -41,10 +41,10 @@ int32_t qwBuildAndSendQueryRsp(SQWConnInfo *pConn, int32_t code);
int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo, int32_t num); int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo, int32_t num);
void qwFreeFetchRsp(void *msg); void qwFreeFetchRsp(void *msg);
int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp); int32_t qwMallocFetchRsp(int32_t length, SRetrieveTableRsp **rsp);
int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp); int32_t qwGetSchTasksStatus(SQWorker *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp);
int32_t qwBuildAndSendHbRsp(SQWConnInfo *pConn, SSchedulerHbRsp *rsp, int32_t code); int32_t qwBuildAndSendHbRsp(SQWConnInfo *pConn, SSchedulerHbRsp *rsp, int32_t code);
int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn); int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn);
int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo *pConn); int32_t qwRegisterHbBrokenLinkArg(SQWorker *mgmt, uint64_t sId, SQWConnInfo *pConn);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/libs/qworker/src/qworker.c
浏览文件 @ 54fe93f7
...@@ -10,6 +10,11 @@ ...@@ -10,6 +10,11 @@
#include "tname.h" #include "tname.h"
SQWDebug gQWDebug = {.statusEnable = true, .dumpEnable = true}; SQWDebug gQWDebug = {.statusEnable = true, .dumpEnable = true};
SQWorkerMgmt gQwMgmt = {
.lock = 0,
.qwRef = -1,
.qwNum = 0,
};
int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore) { int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore) {
if (!gQWDebug.statusEnable) { if (!gQWDebug.statusEnable) {
...@@ -98,7 +103,7 @@ _return: ...@@ -98,7 +103,7 @@ _return:
void qwDbgDumpSchInfo(SQWSchStatus *sch, int32_t i) {} void qwDbgDumpSchInfo(SQWSchStatus *sch, int32_t i) {}
void qwDbgDumpMgmtInfo(SQWorkerMgmt *mgmt) { void qwDbgDumpMgmtInfo(SQWorker *mgmt) {
if (!gQWDebug.dumpEnable) { if (!gQWDebug.dumpEnable) {
return; return;
} }
...@@ -186,7 +191,7 @@ int32_t qwSetTaskStatus(QW_FPARAMS_DEF, SQWTaskStatus *task, int8_t status) { ...@@ -186,7 +191,7 @@ int32_t qwSetTaskStatus(QW_FPARAMS_DEF, SQWTaskStatus *task, int8_t status) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwAddSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType) { int32_t qwAddSchedulerImpl(SQWorker *mgmt, uint64_t sId, int32_t rwType) {
SQWSchStatus newSch = {0}; SQWSchStatus newSch = {0};
newSch.tasksHash = newSch.tasksHash =
taosHashInit(mgmt->cfg.maxSchTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK); taosHashInit(mgmt->cfg.maxSchTaskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
...@@ -213,7 +218,7 @@ int32_t qwAddSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType) { ...@@ -213,7 +218,7 @@ int32_t qwAddSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwAcquireSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch, int32_t nOpt) { int32_t qwAcquireSchedulerImpl(SQWorker *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch, int32_t nOpt) {
while (true) { while (true) {
QW_LOCK(rwType, &mgmt->schLock); QW_LOCK(rwType, &mgmt->schLock);
*sch = taosHashGet(mgmt->schHash, &sId, sizeof(sId)); *sch = taosHashGet(mgmt->schHash, &sId, sizeof(sId));
...@@ -240,15 +245,15 @@ int32_t qwAcquireSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType, ...@@ -240,15 +245,15 @@ int32_t qwAcquireSchedulerImpl(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType,
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwAcquireAddScheduler(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch) { int32_t qwAcquireAddScheduler(SQWorker *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch) {
return qwAcquireSchedulerImpl(mgmt, sId, rwType, sch, QW_NOT_EXIST_ADD); return qwAcquireSchedulerImpl(mgmt, sId, rwType, sch, QW_NOT_EXIST_ADD);
} }
int32_t qwAcquireScheduler(SQWorkerMgmt *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch) { int32_t qwAcquireScheduler(SQWorker *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch) {
return qwAcquireSchedulerImpl(mgmt, sId, rwType, sch, QW_NOT_EXIST_RET_ERR); return qwAcquireSchedulerImpl(mgmt, sId, rwType, sch, QW_NOT_EXIST_RET_ERR);
} }
void qwReleaseScheduler(int32_t rwType, SQWorkerMgmt *mgmt) { QW_UNLOCK(rwType, &mgmt->schLock); } void qwReleaseScheduler(int32_t rwType, SQWorker *mgmt) { QW_UNLOCK(rwType, &mgmt->schLock); }
int32_t qwAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, SQWTaskStatus **task) { int32_t qwAcquireTaskStatus(QW_FPARAMS_DEF, int32_t rwType, SQWSchStatus *sch, SQWTaskStatus **task) {
char id[sizeof(qId) + sizeof(tId)] = {0}; char id[sizeof(qId) + sizeof(tId)] = {0};
...@@ -384,7 +389,7 @@ int32_t qwAddTaskCtx(QW_FPARAMS_DEF) { QW_RET(qwAddTaskCtxImpl(QW_FPARAMS(), fal ...@@ -384,7 +389,7 @@ int32_t qwAddTaskCtx(QW_FPARAMS_DEF) { QW_RET(qwAddTaskCtxImpl(QW_FPARAMS(), fal
int32_t qwAddAcquireTaskCtx(QW_FPARAMS_DEF, SQWTaskCtx **ctx) { return qwAddTaskCtxImpl(QW_FPARAMS(), true, ctx); } int32_t qwAddAcquireTaskCtx(QW_FPARAMS_DEF, SQWTaskCtx **ctx) { return qwAddTaskCtxImpl(QW_FPARAMS(), true, ctx); }
void qwReleaseTaskCtx(SQWorkerMgmt *mgmt, void *ctx) { taosHashRelease(mgmt->ctxHash, ctx); } void qwReleaseTaskCtx(SQWorker *mgmt, void *ctx) { taosHashRelease(mgmt->ctxHash, ctx); }
void qwFreeTaskHandle(QW_FPARAMS_DEF, qTaskInfo_t *taskHandle) { void qwFreeTaskHandle(QW_FPARAMS_DEF, qTaskInfo_t *taskHandle) {
// Note: free/kill may in RC // Note: free/kill may in RC
...@@ -606,7 +611,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) { ...@@ -606,7 +611,7 @@ int32_t qwExecTask(QW_FPARAMS_DEF, SQWTaskCtx *ctx, bool *queryEnd) {
QW_RET(code); QW_RET(code);
} }
int32_t qwGenerateSchHbRsp(SQWorkerMgmt *mgmt, SQWSchStatus *sch, SQWHbInfo *hbInfo) { int32_t qwGenerateSchHbRsp(SQWorker *mgmt, SQWSchStatus *sch, SQWHbInfo *hbInfo) {
int32_t taskNum = 0; int32_t taskNum = 0;
hbInfo->connInfo = sch->hbConnInfo; hbInfo->connInfo = sch->hbConnInfo;
...@@ -1262,7 +1267,7 @@ _return: ...@@ -1262,7 +1267,7 @@ _return:
QW_RET(TSDB_CODE_SUCCESS); QW_RET(TSDB_CODE_SUCCESS);
} }
int32_t qwProcessHbLinkBroken(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) { int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
int32_t code = 0; int32_t code = 0;
SSchedulerHbRsp rsp = {0}; SSchedulerHbRsp rsp = {0};
SQWSchStatus * sch = NULL; SQWSchStatus * sch = NULL;
...@@ -1288,7 +1293,7 @@ int32_t qwProcessHbLinkBroken(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq ...@@ -1288,7 +1293,7 @@ int32_t qwProcessHbLinkBroken(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq
QW_RET(TSDB_CODE_SUCCESS); QW_RET(TSDB_CODE_SUCCESS);
} }
int32_t qwProcessHb(SQWorkerMgmt *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) { int32_t qwProcessHb(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *req) {
int32_t code = 0; int32_t code = 0;
SSchedulerHbRsp rsp = {0}; SSchedulerHbRsp rsp = {0};
SQWSchStatus * sch = NULL; SQWSchStatus * sch = NULL;
...@@ -1333,7 +1338,20 @@ _return: ...@@ -1333,7 +1338,20 @@ _return:
} }
void qwProcessHbTimerEvent(void *param, void *tmrId) { void qwProcessHbTimerEvent(void *param, void *tmrId) {
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)param; SQWHbParam* hbParam = (SQWHbParam*)param;
if (hbParam->qwrId != atomic_load_32(&gQwMgmt.qwRef)) {
taosMemoryFree(param);
return;
}
int64_t refId = hbParam->refId;
SQWorker *mgmt = qwAcquire(refId);
if (NULL == mgmt) {
QW_DLOG("qwAcquire %" PRIx64 "failed", refId);
taosMemoryFree(param);
return;
}
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
int32_t taskNum = 0; int32_t taskNum = 0;
SQWHbInfo * rspList = NULL; SQWHbInfo * rspList = NULL;
...@@ -1347,6 +1365,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { ...@@ -1347,6 +1365,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
if (schNum <= 0) { if (schNum <= 0) {
QW_UNLOCK(QW_READ, &mgmt->schLock); QW_UNLOCK(QW_READ, &mgmt->schLock);
taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
qwRelease(refId);
return; return;
} }
...@@ -1355,6 +1374,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { ...@@ -1355,6 +1374,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
QW_UNLOCK(QW_READ, &mgmt->schLock); QW_UNLOCK(QW_READ, &mgmt->schLock);
QW_ELOG("calloc %d SQWHbInfo failed", schNum); QW_ELOG("calloc %d SQWHbInfo failed", schNum);
taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
qwRelease(refId);
return; return;
} }
...@@ -1396,18 +1416,72 @@ _return: ...@@ -1396,18 +1416,72 @@ _return:
taosMemoryFreeClear(rspList); taosMemoryFreeClear(rspList);
taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
qwRelease(refId);
}
void qwCloseRef(void) {
taosWLockLatch(&gQwMgmt.lock);
if (atomic_load_32(&gQwMgmt.qwNum) <= 0 && gQwMgmt.qwRef >= 0) {
taosCloseRef(gQwMgmt.qwRef);
gQwMgmt.qwRef= -1;
}
taosWUnLockLatch(&gQwMgmt.lock);
} }
void qwDestroyImpl(void *pMgmt) {
SQWorker *mgmt = (SQWorker *)pMgmt;
taosTmrStopA(&mgmt->hbTimer);
taosTmrCleanUp(mgmt->timer);
// TODO STOP ALL QUERY
// TODO FREE ALL
taosHashCleanup(mgmt->ctxHash);
taosHashCleanup(mgmt->schHash);
taosMemoryFree(mgmt);
atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
qwCloseRef();
}
int32_t qwOpenRef(void) {
taosWLockLatch(&gQwMgmt.lock);
if (gQwMgmt.qwRef < 0) {
gQwMgmt.qwRef= taosOpenRef(100, qwDestroyImpl);
if (gQwMgmt.qwRef < 0) {
taosWUnLockLatch(&gQwMgmt.lock);
qError("init qworker ref failed");
QW_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
}
taosWUnLockLatch(&gQwMgmt.lock);
return TSDB_CODE_SUCCESS;
}
int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, const SMsgCb *pMsgCb) { int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, const SMsgCb *pMsgCb) {
if (NULL == qWorkerMgmt || pMsgCb->pWrapper == NULL) { if (NULL == qWorkerMgmt || pMsgCb->pWrapper == NULL) {
qError("invalid param to init qworker"); qError("invalid param to init qworker");
QW_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
int32_t code = 0; atomic_add_fetch_32(&gQwMgmt.qwNum, 1);
SQWorkerMgmt *mgmt = taosMemoryCalloc(1, sizeof(SQWorkerMgmt));
int32_t code = qwOpenRef();
if (code) {
atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
QW_RET(code);
}
SQWorker *mgmt = taosMemoryCalloc(1, sizeof(SQWorker));
if (NULL == mgmt) { if (NULL == mgmt) {
qError("calloc %d failed", (int32_t)sizeof(SQWorkerMgmt)); qError("calloc %d failed", (int32_t)sizeof(SQWorker));
atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
QW_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
...@@ -1449,16 +1523,30 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW ...@@ -1449,16 +1523,30 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW
QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY); QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
mgmt->hbTimer = taosTmrStart(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, mgmt, mgmt->timer);
if (NULL == mgmt->hbTimer) {
qError("start hb timer failed");
QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
mgmt->nodeType = nodeType; mgmt->nodeType = nodeType;
mgmt->nodeId = nodeId; mgmt->nodeId = nodeId;
mgmt->msgCb = *pMsgCb; mgmt->msgCb = *pMsgCb;
mgmt->refId = taosAddRef(gQwMgmt.qwRef, mgmt);
if (mgmt->refId < 0) {
qError("taosAddRef qw failed, error:%s", tstrerror(terrno));
QW_ERR_JRET(terrno);
}
SQWHbParam *param = taosMemoryMalloc(sizeof(SQWHbParam));
if (NULL == param) {
qError("malloc hb param failed, error:%s", tstrerror(terrno));
QW_ERR_JRET(terrno);
}
param->qwrId = gQwMgmt.qwRef;
param->refId = mgmt->refId;
mgmt->hbTimer = taosTmrStart(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, (void*)param, mgmt->timer);
if (NULL == mgmt->hbTimer) {
qError("start hb timer failed");
QW_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
}
*qWorkerMgmt = mgmt; *qWorkerMgmt = mgmt;
qDebug("qworker initialized for node, type:%d, id:%d, handle:%p", mgmt->nodeType, mgmt->nodeId, mgmt); qDebug("qworker initialized for node, type:%d, id:%d, handle:%p", mgmt->nodeType, mgmt->nodeId, mgmt);
...@@ -1467,13 +1555,17 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW ...@@ -1467,13 +1555,17 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW
_return: _return:
taosHashCleanup(mgmt->schHash); if (mgmt->refId >= 0) {
taosHashCleanup(mgmt->ctxHash); qwRelease(mgmt->refId);
} else {
taosTmrCleanUp(mgmt->timer); taosHashCleanup(mgmt->schHash);
taosHashCleanup(mgmt->ctxHash);
taosMemoryFreeClear(mgmt); taosTmrCleanUp(mgmt->timer);
taosMemoryFreeClear(mgmt);
atomic_sub_fetch_32(&gQwMgmt.qwNum, 1);
}
QW_RET(code); QW_RET(code);
} }
...@@ -1482,22 +1574,14 @@ void qWorkerDestroy(void **qWorkerMgmt) { ...@@ -1482,22 +1574,14 @@ void qWorkerDestroy(void **qWorkerMgmt) {
return; return;
} }
SQWorkerMgmt *mgmt = *qWorkerMgmt; SQWorker *mgmt = *qWorkerMgmt;
taosTmrStopA(&mgmt->hbTimer);
taosTmrCleanUp(mgmt->timer);
// TODO STOP ALL QUERY
// TODO FREE ALL
taosHashCleanup(mgmt->ctxHash);
taosHashCleanup(mgmt->schHash);
taosMemoryFreeClear(*qWorkerMgmt); if (taosRemoveRef(gQwMgmt.qwRef, mgmt->refId)) {
qError("remove qw from ref list failed, refId:%" PRIx64, mgmt->refId);
}
} }
int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp) { int32_t qwGetSchTasksStatus(SQWorker *mgmt, uint64_t sId, SSchedulerStatusRsp **rsp) {
/* /*
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
int32_t taskNum = 0; int32_t taskNum = 0;
...@@ -1544,7 +1628,7 @@ int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRs ...@@ -1544,7 +1628,7 @@ int32_t qwGetSchTasksStatus(SQWorkerMgmt *mgmt, uint64_t sId, SSchedulerStatusRs
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwUpdateSchLastAccess(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) { int32_t qwUpdateSchLastAccess(SQWorker *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) {
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
/* /*
...@@ -1557,7 +1641,7 @@ int32_t qwUpdateSchLastAccess(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, ui ...@@ -1557,7 +1641,7 @@ int32_t qwUpdateSchLastAccess(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, ui
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int8_t *taskStatus) { int32_t qwGetTaskStatus(SQWorker *mgmt, uint64_t sId, uint64_t qId, uint64_t tId, int8_t *taskStatus) {
SQWSchStatus * sch = NULL; SQWSchStatus * sch = NULL;
SQWTaskStatus *task = NULL; SQWTaskStatus *task = NULL;
int32_t code = 0; int32_t code = 0;
...@@ -1584,7 +1668,7 @@ int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t ...@@ -1584,7 +1668,7 @@ int32_t qwGetTaskStatus(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t
QW_RET(code); QW_RET(code);
} }
int32_t qwCancelTask(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) { int32_t qwCancelTask(SQWorker *mgmt, uint64_t sId, uint64_t qId, uint64_t tId) {
SQWSchStatus * sch = NULL; SQWSchStatus * sch = NULL;
SQWTaskStatus *task = NULL; SQWTaskStatus *task = NULL;
int32_t code = 0; int32_t code = 0;
......
source/libs/qworker/src/qworkerMsg.c
浏览文件 @ 54fe93f7
...@@ -319,7 +319,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) { ...@@ -319,7 +319,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SQWConnInfo *pConn) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t qwRegisterHbBrokenLinkArg(SQWorkerMgmt *mgmt, uint64_t sId, SQWConnInfo *pConn) { int32_t qwRegisterHbBrokenLinkArg(SQWorker *mgmt, uint64_t sId, SQWConnInfo *pConn) {
SSchedulerHbReq req = {0}; SSchedulerHbReq req = {0};
req.header.vgId = mgmt->nodeId; req.header.vgId = mgmt->nodeId;
req.sId = sId; req.sId = sId;
...@@ -362,7 +362,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -362,7 +362,7 @@ int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
int32_t code = 0; int32_t code = 0;
SSubQueryMsg *msg = pMsg->pCont; SSubQueryMsg *msg = pMsg->pCont;
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
if (NULL == msg || pMsg->contLen <= sizeof(*msg)) { if (NULL == msg || pMsg->contLen <= sizeof(*msg)) {
QW_ELOG("invalid query msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("invalid query msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
...@@ -404,7 +404,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -404,7 +404,7 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont; SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont;
bool needStop = false; bool needStop = false;
SQWTaskCtx *handles = NULL; SQWTaskCtx *handles = NULL;
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
...@@ -435,7 +435,7 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -435,7 +435,7 @@ int32_t qWorkerProcessReadyMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT; return TSDB_CODE_QRY_INVALID_INPUT;
} }
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
SResReadyReq *msg = pMsg->pCont; SResReadyReq *msg = pMsg->pCont;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
QW_ELOG("invalid task ready msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("invalid task ready msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
...@@ -477,7 +477,7 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -477,7 +477,7 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); QW_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
} }
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
msg->sId = htobe64(msg->sId); msg->sId = htobe64(msg->sId);
uint64_t sId = msg->sId; uint64_t sId = msg->sId;
...@@ -498,7 +498,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -498,7 +498,7 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
} }
SResFetchReq *msg = pMsg->pCont; SResFetchReq *msg = pMsg->pCont;
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
QW_ELOG("invalid fetch msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("invalid fetch msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
...@@ -538,7 +538,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -538,7 +538,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
return TSDB_CODE_QRY_INVALID_INPUT; return TSDB_CODE_QRY_INVALID_INPUT;
} }
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
int32_t code = 0; int32_t code = 0;
STaskCancelReq *msg = pMsg->pCont; STaskCancelReq *msg = pMsg->pCont;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
...@@ -578,7 +578,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -578,7 +578,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
int32_t code = 0; int32_t code = 0;
STaskDropReq *msg = pMsg->pCont; STaskDropReq *msg = pMsg->pCont;
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
if (NULL == msg || pMsg->contLen < sizeof(*msg)) { if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
QW_ELOG("invalid task drop msg, msg:%p, msgLen:%d", msg, pMsg->contLen); QW_ELOG("invalid task drop msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
...@@ -620,7 +620,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) { ...@@ -620,7 +620,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
int32_t code = 0; int32_t code = 0;
SSchedulerHbReq req = {0}; SSchedulerHbReq req = {0};
SQWorkerMgmt *mgmt = (SQWorkerMgmt *)qWorkerMgmt; SQWorker *mgmt = (SQWorker *)qWorkerMgmt;
if (NULL == pMsg->pCont) { if (NULL == pMsg->pCont) {
QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen); QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen);
......
tests/script/api/batchprepare.c
浏览文件 @ 54fe93f7
...@@ -188,8 +188,6 @@ CaseCtrl gCaseCtrl = { ...@@ -188,8 +188,6 @@ CaseCtrl gCaseCtrl = {
.caseRunIdx = -1, .caseRunIdx = -1,
// .caseRunNum = -1, // .caseRunNum = -1,
.bindColTypeNum = tListLen(bindColTypeList),
.bindColTypeList = bindColTypeList,
.caseIdx = 22, .caseIdx = 22,
.caseNum = 1, .caseNum = 1,
.caseRunNum = 1, .caseRunNum = 1,
...@@ -318,7 +316,7 @@ void generateInsertSQL(BindData *data) { ...@@ -318,7 +316,7 @@ void generateInsertSQL(BindData *data) {
len += sprintf(data->sql + len, "ubigdata"); len += sprintf(data->sql + len, "ubigdata");
break; break;
default: default:
printf("invalid col type:%d", data->pBind[c].buffer_type); printf("!!!invalid col type:%d", data->pBind[c].buffer_type);
exit(1); exit(1);
} }
} }
...@@ -336,7 +334,7 @@ void generateInsertSQL(BindData *data) { ...@@ -336,7 +334,7 @@ void generateInsertSQL(BindData *data) {
len += sprintf(data->sql + len, ")"); len += sprintf(data->sql + len, ")");
if (gCaseCtrl.printStmtSql) { if (gCaseCtrl.printStmtSql) {
printf("SQL: %s\n", data->sql); printf("\tSQL: %s\n", data->sql);
} }
} }
...@@ -358,7 +356,7 @@ void bpAppendOperatorParam(BindData *data, int32_t *len, int32_t dataType) { ...@@ -358,7 +356,7 @@ void bpAppendOperatorParam(BindData *data, int32_t *len, int32_t dataType) {
} }
break; break;
default: default:
printf("invalid paramNum:%d\n", pInfo->paramNum); printf("!!!invalid paramNum:%d\n", pInfo->paramNum);
exit(1); exit(1);
} }
} }
...@@ -414,7 +412,7 @@ void generateQuerySQL(BindData *data, int32_t tblIdx) { ...@@ -414,7 +412,7 @@ void generateQuerySQL(BindData *data, int32_t tblIdx) {
len += sprintf(data->sql + len, "ubigdata"); len += sprintf(data->sql + len, "ubigdata");
break; break;
default: default:
printf("invalid col type:%d", data->pBind[c].buffer_type); printf("!!!invalid col type:%d", data->pBind[c].buffer_type);
exit(1); exit(1);
} }
...@@ -423,7 +421,7 @@ void generateQuerySQL(BindData *data, int32_t tblIdx) { ...@@ -423,7 +421,7 @@ void generateQuerySQL(BindData *data, int32_t tblIdx) {
} }
if (gCaseCtrl.printStmtSql) { if (gCaseCtrl.printStmtSql) {
printf("SQL: %s\n", data->sql); printf("\tSTMT SQL: %s\n", data->sql);
} }
} }
...@@ -551,7 +549,7 @@ int32_t prepareColData(BindData *data, int32_t bindIdx, int32_t rowIdx, int32_t ...@@ -551,7 +549,7 @@ int32_t prepareColData(BindData *data, int32_t bindIdx, int32_t rowIdx, int32_t
data->pBind[bindIdx].is_null = data->isNull ? (data->isNull + rowIdx) : NULL; data->pBind[bindIdx].is_null = data->isNull ? (data->isNull + rowIdx) : NULL;
break; break;
default: default:
printf("invalid col type:%d", dataType); printf("!!!invalid col type:%d", dataType);
exit(1); exit(1);
} }
...@@ -709,7 +707,7 @@ void bpFetchRows(TAOS_RES *result, bool printr, int32_t *rows) { ...@@ -709,7 +707,7 @@ void bpFetchRows(TAOS_RES *result, bool printr, int32_t *rows) {
if (printr) { if (printr) {
memset(temp, 0, sizeof(temp)); memset(temp, 0, sizeof(temp));
taos_print_row(temp, row, fields, num_fields); taos_print_row(temp, row, fields, num_fields);
printf("[%s]\n", temp); printf("\t[%s]\n", temp);
} }
} }
} }
...@@ -718,7 +716,7 @@ void bpExecQuery(TAOS * taos, char* sql, bool printr, int32_t *rows) { ...@@ -718,7 +716,7 @@ void bpExecQuery(TAOS * taos, char* sql, bool printr, int32_t *rows) {
TAOS_RES *result = taos_query(taos, sql); TAOS_RES *result = taos_query(taos, sql);
int code = taos_errno(result); int code = taos_errno(result);
if (code != 0) { if (code != 0) {
printf("failed to query table, reason:%s\n", taos_errstr(result)); printf("!!!failed to query table, reason:%s\n", taos_errstr(result));
taos_free_result(result); taos_free_result(result);
exit(1); exit(1);
} }
...@@ -791,7 +789,7 @@ int32_t bpAppendValueString(char *buf, int type, void *value, int32_t valueLen, ...@@ -791,7 +789,7 @@ int32_t bpAppendValueString(char *buf, int type, void *value, int32_t valueLen,
break; break;
default: default:
printf("invalid data type:%d\n", type); printf("!!!invalid data type:%d\n", type);
exit(1); exit(1);
} }
} }
...@@ -803,13 +801,13 @@ int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) { ...@@ -803,13 +801,13 @@ int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
if (gCurCase->bindRowNum > 1) { if (gCurCase->bindRowNum > 1) {
if (0 == (n++%2)) { if (0 == (n++%2)) {
if (taos_stmt_bind_param_batch(stmt, bind)) { if (taos_stmt_bind_param_batch(stmt, bind)) {
printf("taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} else { } else {
for (int32_t i = 0; i < gCurCase->bindColNum; ++i) { for (int32_t i = 0; i < gCurCase->bindColNum; ++i) {
if (taos_stmt_bind_single_param_batch(stmt, bind++, i)) { if (taos_stmt_bind_single_param_batch(stmt, bind++, i)) {
printf("taos_stmt_bind_single_param_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_bind_single_param_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -817,12 +815,12 @@ int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) { ...@@ -817,12 +815,12 @@ int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
} else { } else {
if (0 == (n++%2)) { if (0 == (n++%2)) {
if (taos_stmt_bind_param_batch(stmt, bind)) { if (taos_stmt_bind_param_batch(stmt, bind)) {
printf("taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_bind_param_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} else { } else {
if (taos_stmt_bind_param(stmt, bind)) { if (taos_stmt_bind_param(stmt, bind)) {
printf("taos_stmt_bind_param error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_bind_param error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -834,12 +832,12 @@ int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) { ...@@ -834,12 +832,12 @@ int32_t bpBindParam(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind) {
void bpCheckIsInsert(TAOS_STMT *stmt, int32_t insert) { void bpCheckIsInsert(TAOS_STMT *stmt, int32_t insert) {
int32_t isInsert = 0; int32_t isInsert = 0;
if (taos_stmt_is_insert(stmt, &isInsert)) { if (taos_stmt_is_insert(stmt, &isInsert)) {
printf("taos_stmt_is_insert error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_is_insert error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
if (insert != isInsert) { if (insert != isInsert) {
printf("is insert failed\n"); printf("!!!is insert failed\n");
exit(1); exit(1);
} }
} }
...@@ -847,12 +845,12 @@ void bpCheckIsInsert(TAOS_STMT *stmt, int32_t insert) { ...@@ -847,12 +845,12 @@ void bpCheckIsInsert(TAOS_STMT *stmt, int32_t insert) {
void bpCheckParamNum(TAOS_STMT *stmt) { void bpCheckParamNum(TAOS_STMT *stmt) {
int32_t num = 0; int32_t num = 0;
if (taos_stmt_num_params(stmt, &num)) { if (taos_stmt_num_params(stmt, &num)) {
printf("taos_stmt_num_params error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_num_params error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
if (gCurCase->bindColNum != num) { if (gCurCase->bindColNum != num) {
printf("is insert failed\n"); printf("!!!is insert failed\n");
exit(1); exit(1);
} }
} }
...@@ -861,7 +859,7 @@ void bpCheckAffectedRows(TAOS_STMT *stmt, int32_t times) { ...@@ -861,7 +859,7 @@ void bpCheckAffectedRows(TAOS_STMT *stmt, int32_t times) {
int32_t rows = taos_stmt_affected_rows(stmt); int32_t rows = taos_stmt_affected_rows(stmt);
int32_t insertNum = gCurCase->rowNum * gCurCase->tblNum * times; int32_t insertNum = gCurCase->rowNum * gCurCase->tblNum * times;
if (insertNum != rows) { if (insertNum != rows) {
printf("affected rows %d mis-match with insert num %d\n", rows, insertNum); printf("!!!affected rows %d mis-match with insert num %d\n", rows, insertNum);
exit(1); exit(1);
} }
} }
...@@ -869,7 +867,7 @@ void bpCheckAffectedRows(TAOS_STMT *stmt, int32_t times) { ...@@ -869,7 +867,7 @@ void bpCheckAffectedRows(TAOS_STMT *stmt, int32_t times) {
void bpCheckAffectedRowsOnce(TAOS_STMT *stmt, int32_t expectedNum) { void bpCheckAffectedRowsOnce(TAOS_STMT *stmt, int32_t expectedNum) {
int32_t rows = taos_stmt_affected_rows_once(stmt); int32_t rows = taos_stmt_affected_rows_once(stmt);
if (expectedNum != rows) { if (expectedNum != rows) {
printf("affected rows %d mis-match with expected num %d\n", rows, expectedNum); printf("!!!affected rows %d mis-match with expected num %d\n", rows, expectedNum);
exit(1); exit(1);
} }
} }
...@@ -904,16 +902,16 @@ void bpCheckQueryResult(TAOS_STMT *stmt, TAOS *taos, char *stmtSql, TAOS_MULTI_B ...@@ -904,16 +902,16 @@ void bpCheckQueryResult(TAOS_STMT *stmt, TAOS *taos, char *stmtSql, TAOS_MULTI_B
} }
if (gCaseCtrl.printQuerySql) { if (gCaseCtrl.printQuerySql) {
printf("Query SQL: %s\n", sql); printf("\tQuery SQL: %s\n", sql);
} }
bpExecQuery(taos, sql, gCaseCtrl.printRes, &sqlResNum); bpExecQuery(taos, sql, gCaseCtrl.printRes, &sqlResNum);
if (sqlResNum != stmtResNum) { if (sqlResNum != stmtResNum) {
printf("sql res num %d mis-match stmt res num %d\n", sqlResNum, stmtResNum); printf("!!!sql res num %d mis-match stmt res num %d\n", sqlResNum, stmtResNum);
exit(1); exit(1);
} }
printf("sql res num match stmt res num %d\n", stmtResNum); printf("***sql res num match stmt res num %d\n", stmtResNum);
} }
/* prepare [settbname [bind add]] exec */ /* prepare [settbname [bind add]] exec */
...@@ -923,7 +921,7 @@ int insertMBSETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -923,7 +921,7 @@ int insertMBSETest1(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -936,7 +934,7 @@ int insertMBSETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -936,7 +934,7 @@ int insertMBSETest1(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -951,14 +949,14 @@ int insertMBSETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -951,14 +949,14 @@ int insertMBSETest1(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -978,7 +976,7 @@ int insertMBSETest2(TAOS_STMT *stmt, TAOS *taos) { ...@@ -978,7 +976,7 @@ int insertMBSETest2(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -993,7 +991,7 @@ int insertMBSETest2(TAOS_STMT *stmt, TAOS *taos) { ...@@ -993,7 +991,7 @@ int insertMBSETest2(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1007,14 +1005,14 @@ int insertMBSETest2(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1007,14 +1005,14 @@ int insertMBSETest2(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1033,7 +1031,7 @@ int insertMBMETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1033,7 +1031,7 @@ int insertMBMETest1(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1046,7 +1044,7 @@ int insertMBMETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1046,7 +1044,7 @@ int insertMBMETest1(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1061,13 +1059,13 @@ int insertMBMETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1061,13 +1059,13 @@ int insertMBMETest1(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1087,7 +1085,7 @@ int insertMBMETest2(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1087,7 +1085,7 @@ int insertMBMETest2(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1100,7 +1098,7 @@ int insertMBMETest2(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1100,7 +1098,7 @@ int insertMBMETest2(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1115,12 +1113,12 @@ int insertMBMETest2(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1115,12 +1113,12 @@ int insertMBMETest2(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1141,7 +1139,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1141,7 +1139,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1154,7 +1152,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1154,7 +1152,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1169,7 +1167,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1169,7 +1167,7 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1179,12 +1177,12 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1179,12 +1177,12 @@ int insertMBMETest3(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1206,7 +1204,7 @@ int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1206,7 +1204,7 @@ int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1221,7 +1219,7 @@ int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1221,7 +1219,7 @@ int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1235,12 +1233,12 @@ int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1235,12 +1233,12 @@ int insertMBMETest4(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1264,7 +1262,7 @@ int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1264,7 +1262,7 @@ int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1277,7 +1275,7 @@ int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1277,7 +1275,7 @@ int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos) {
sprintf(buf, "t%d", t); sprintf(buf, "t%d", t);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_set_tbname error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1292,13 +1290,13 @@ int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1292,13 +1290,13 @@ int insertMPMETest1(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
} }
...@@ -1328,7 +1326,7 @@ int querySUBTTest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1328,7 +1326,7 @@ int querySUBTTest1(TAOS_STMT *stmt, TAOS *taos) {
int code = taos_stmt_prepare(stmt, data.sql, 0); int code = taos_stmt_prepare(stmt, data.sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt)); printf("!!!failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -1344,12 +1342,12 @@ int querySUBTTest1(TAOS_STMT *stmt, TAOS *taos) { ...@@ -1344,12 +1342,12 @@ int querySUBTTest1(TAOS_STMT *stmt, TAOS *taos) {
} }
if (taos_stmt_add_batch(stmt)) { if (taos_stmt_add_batch(stmt)) {
printf("taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_add_batch error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_execute error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
...@@ -4249,10 +4247,10 @@ void prepareCheckResultImpl(TAOS * taos, char *tname, bool printr, int expec ...@@ -4249,10 +4247,10 @@ void prepareCheckResultImpl(TAOS * taos, char *tname, bool printr, int expec
if (rows == expected) { if (rows == expected) {
if (!silent) { if (!silent) {
printf("%d rows are fetched as expected from %s\n", rows, tname); printf("***%d rows are fetched as expected from %s\n", rows, tname);
} }
} else { } else {
printf("!!!expect %d rows, but %d rows are fetched from %s\n", expected, rows, tname); printf("!!!expect rows %d mis-match rows %d fetched from %s\n", expected, rows, tname);
exit(1); exit(1);
} }
} }
...@@ -4302,7 +4300,7 @@ int sql_perf1(TAOS *taos) { ...@@ -4302,7 +4300,7 @@ int sql_perf1(TAOS *taos) {
result = taos_query(taos, sql[i]); result = taos_query(taos, sql[i]);
int code = taos_errno(result); int code = taos_errno(result);
if (code != 0) { if (code != 0) {
printf("failed to query table, reason:%s\n", taos_errstr(result)); printf("%d failed to query table, reason:%s\n", taos_errstr(result));
taos_free_result(result); taos_free_result(result);
exit(1); exit(1);
} }
...@@ -4539,7 +4537,7 @@ void generateCreateTableSQL(char *buf, int32_t tblIdx, int32_t colNum, int32_t * ...@@ -4539,7 +4537,7 @@ void generateCreateTableSQL(char *buf, int32_t tblIdx, int32_t colNum, int32_t *
blen += sprintf(buf + blen, ")"); blen += sprintf(buf + blen, ")");
if (gCaseCtrl.printCreateTblSql) { if (gCaseCtrl.printCreateTblSql) {
printf("Create Table SQL:%s\n", buf); printf("\tCreate Table SQL:%s\n", buf);
} }
} }
...@@ -4553,7 +4551,7 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) { ...@@ -4553,7 +4551,7 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) {
result = taos_query(taos, "create database demo keep 36500"); result = taos_query(taos, "create database demo keep 36500");
code = taos_errno(result); code = taos_errno(result);
if (code != 0) { if (code != 0) {
printf("failed to create database, reason:%s\n", taos_errstr(result)); printf("!!!failed to create database, reason:%s\n", taos_errstr(result));
taos_free_result(result); taos_free_result(result);
exit(1); exit(1);
} }
...@@ -4570,7 +4568,7 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) { ...@@ -4570,7 +4568,7 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) {
result = taos_query(taos, buf); result = taos_query(taos, buf);
code = taos_errno(result); code = taos_errno(result);
if (code != 0) { if (code != 0) {
printf("failed to create table, reason:%s\n", taos_errstr(result)); printf("!!!failed to create table, reason:%s\n", taos_errstr(result));
taos_free_result(result); taos_free_result(result);
exit(1); exit(1);
} }
...@@ -4583,7 +4581,7 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) { ...@@ -4583,7 +4581,7 @@ void prepare(TAOS *taos, int32_t colNum, int32_t *colList, int prepareStb) {
result = taos_query(taos, buf); result = taos_query(taos, buf);
code = taos_errno(result); code = taos_errno(result);
if (code != 0) { if (code != 0) {
printf("failed to create table, reason:%s\n", taos_errstr(result)); printf("!!!failed to create table, reason:%s\n", taos_errstr(result));
taos_free_result(result); taos_free_result(result);
exit(1); exit(1);
} }
...@@ -4654,7 +4652,7 @@ int32_t runCase(TAOS *taos, int32_t caseIdx, int32_t caseRunIdx, bool silent) { ...@@ -4654,7 +4652,7 @@ int32_t runCase(TAOS *taos, int32_t caseIdx, int32_t caseRunIdx, bool silent) {
stmt = taos_stmt_init(taos); stmt = taos_stmt_init(taos);
if (NULL == stmt) { if (NULL == stmt) {
printf("taos_stmt_init failed, error:%s\n", taos_stmt_errstr(stmt)); printf("!!!taos_stmt_init failed, error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或