xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #2253 from taosdata/feature/query 

Feature/query
This commit is contained in:
Shengliang Guan 2020-06-12 11:21:16 +08:00 committed by GitHub
parent e681737667 df41013eca
commit 1f1ae67076
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 96 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/client/src/tscServer.c
View File

@ -209,6 +209,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
tscError("%p sql is already released", pSql->signature);
return;
}
if (pSql->signature != pSql) {
tscError("%p sql is already released, signature:%p", pSql, pSql->signature);
return;
@ -217,10 +218,9 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcIpSet *pIpSet) {
SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd;
STscObj *pObj = pSql->pTscObj;
// tscTrace("%p msg:%s is received from server", pSql, taosMsg[rpcMsg->msgType]);
if (pObj->signature != pObj) {
tscTrace("%p sql is already released or DB connection is closed, freed:%d pObj:%p signature:%p", pSql, pSql->freed,
if (pObj->signature != pObj || pSql->freed == 1) {
tscTrace("%p sqlObj needs to be released or DB connection is closed, freed:%d pObj:%p signature:%p", pSql, pSql->freed,
pObj, pObj->signature);
tscFreeSqlObj(pSql);
rpcFreeCont(rpcMsg->pCont);
@ -375,7 +375,7 @@ int tscProcessSql(SSqlObj *pSql) {
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex);
STableMetaInfo *pTableMetaInfo = NULL;
uint16_t type = 0;
uint32_t type = 0;
if (pQueryInfo != NULL) {
pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0);
@ -424,7 +424,7 @@ void tscKillSTableQuery(SSqlObj *pSql) {
* sub-queries not correctly released and master sql object of super table query reaches an abnormal state.
*/
pSql->pSubs[i]->res.code = TSDB_CODE_TSC_QUERY_CANCELLED;
//taosStopRpcConn(pSql->pSubs[i]->thandle);
// taosStopRpcConn(pSql->pSubs[i]->);
}
/*

35
src/client/src/tscSql.c
View File

@ -219,6 +219,11 @@ void waitForQueryRsp(void *param, TAOS_RES *tres, int code) {
sem_post(&pSql->rspSem);
}
static void waitForRetrieveRsp(void *param, TAOS_RES *tres, int numOfRows) {
SSqlObj* pSql = (SSqlObj*) tres;
sem_post(&pSql->rspSem);
}
TAOS_RES* taos_query(TAOS *taos, const char *sqlstr) {
STscObj *pObj = (STscObj *)taos;
if (pObj == NULL || pObj->signature != pObj) {
@ -369,11 +374,6 @@ int taos_fetch_block_impl(TAOS_RES *res, TAOS_ROW *rows) {
return (pQueryInfo->order.order == TSDB_ORDER_DESC) ? pRes->numOfRows : -pRes->numOfRows;
}
static void waitForRetrieveRsp(void *param, TAOS_RES *tres, int numOfRows) {
SSqlObj* pSql = (SSqlObj*) tres;
sem_post(&pSql->rspSem);
}
TAOS_ROW taos_fetch_row(TAOS_RES *res) {
SSqlObj *pSql = (SSqlObj *)res;
if (pSql == NULL || pSql->signature != pSql) {
@ -476,7 +476,7 @@ int taos_select_db(TAOS *taos, const char *db) {
}
// send free message to vnode to free qhandle and corresponding resources in vnode
static void tscFreeQhandleInVnode(SSqlObj* pSql) {
static bool tscFreeQhandleInVnode(SSqlObj* pSql) {
SSqlCmd* pCmd = &pSql->cmd;
SSqlRes* pRes = &pSql->res;
@ -496,10 +496,19 @@ static void tscFreeQhandleInVnode(SSqlObj* pSql) {
tscProcessSql(pSql);
// in case of sync model query, waits for response and then goes on
if (pSql->fp == waitForQueryRsp || pSql->fp == waitForRetrieveRsp) {
sem_wait(&pSql->rspSem);
}
// if (pSql->fp == waitForQueryRsp || pSql->fp == waitForRetrieveRsp) {
// sem_wait(&pSql->rspSem);
// tscFreeSqlObj(pSql);
// tscTrace("%p sqlObj is freed by app", pSql);
// } else {
tscTrace("%p sqlObj will be freed while rsp received", pSql);
// }
return true;
}
return false;
}
void taos_free_result(TAOS_RES *res) {
@ -527,10 +536,10 @@ void taos_free_result(TAOS_RES *res) {
}
pQueryInfo->type = TSDB_QUERY_TYPE_FREE_RESOURCE;
tscFreeQhandleInVnode(pSql);
tscFreeSqlObj(pSql);
tscTrace("%p sql result is freed by app", pSql);
if (!tscFreeQhandleInVnode(pSql)) {
tscFreeSqlObj(pSql);
tscTrace("%p sqlObj is freed by app", pSql);
}
}
// todo should not be used in async query

21
src/client/src/tscUtil.c
View File

@ -134,24 +134,6 @@ void tscGetDBInfoFromMeterId(char* tableId, char* db) {
db[0] = 0;
}
//STableIdInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
// if (pSidList == NULL) {
// tscError("illegal sidlist");
// return 0;
// }
//
// if (idx < 0 || idx >= pSidList->numOfSids) {
// int32_t sidRange = (pSidList->numOfSids > 0) ? (pSidList->numOfSids - 1) : 0;
//
// tscError("illegal sidIdx:%d, reset to 0, sidIdx range:%d-%d", idx, 0, sidRange);
// idx = 0;
// }
//
// assert(pSidList->pSidExtInfoList[idx] >= 0);
//
// return (STableIdInfo*)(pSidList->pSidExtInfoList[idx] + (char*)pSidList);
//}
bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
if (pQueryInfo == NULL) {
return false;
@ -176,8 +158,7 @@ bool tscIsTwoStageSTableQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
return false;
}
if (((pQueryInfo->type & TSDB_QUERY_TYPE_STABLE_SUBQUERY) != TSDB_QUERY_TYPE_STABLE_SUBQUERY) &&
pQueryInfo->command == TSDB_SQL_SELECT) {
if (!TSDB_QUERY_HAS_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_STABLE_SUBQUERY) && pQueryInfo->command == TSDB_SQL_SELECT) {
return UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo);
}

2
src/query/inc/tsqlfunction.h
View File

@ -112,7 +112,7 @@ enum {
#define QUERY_IS_STABLE_QUERY(type) (((type)&TSDB_QUERY_TYPE_STABLE_QUERY) != 0)
#define QUERY_IS_JOIN_QUERY(type) (TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_JOIN_QUERY))
#define QUERY_IS_PROJECTION_QUERY(type) (((type)&TSDB_QUERY_TYPE_PROJECTION_QUERY) != 0)
#define QUERY_IS_PROJECTION_QUERY(type) (((type)&TSDB_QUERY_TYPE_PROJECTION_QUERY) != 0)
#define QUERY_IS_FREE_RESOURCE(type) (((type)&TSDB_QUERY_TYPE_FREE_RESOURCE) != 0)
typedef struct SArithmeticSupport {

6
src/query/src/qExecutor.c
View File

@ -4470,7 +4470,7 @@ static void multiTableQueryProcess(SQInfo *pQInfo) {
// query error occurred or query is killed, abort current execution
if (pQInfo->code != TSDB_CODE_SUCCESS || isQueryKilled(pQInfo)) {
qTrace("QInfo:%p query killed or error occurred, code:%d, abort", pQInfo, pQInfo->code);
qTrace("QInfo:%p query killed or error occurred, code:%s, abort", pQInfo, tstrerror(pQInfo->code));
return;
}
@ -4491,7 +4491,7 @@ static void multiTableQueryProcess(SQInfo *pQInfo) {
setQueryStatus(pQuery, QUERY_COMPLETED);
if (pQInfo->code != TSDB_CODE_SUCCESS || isQueryKilled(pQInfo)) {
qTrace("QInfo:%p query killed or error occurred, code:%d, abort", pQInfo, pQInfo->code);
qTrace("QInfo:%p query killed or error occurred, code:%s, abort", pQInfo, tstrerror(pQInfo->code));
return;
}
@ -5851,6 +5851,8 @@ void qDestroyQueryInfo(qinfo_t qHandle) {
}
int16_t ref = T_REF_DEC(pQInfo);
qTrace("QInfo:%p dec refCount, value:%d", pQInfo, ref);
if (ref == 0) {
doDestoryQueryInfo(pQInfo);
}

1
src/rpc/src/rpcMain.c
View File

@ -393,7 +393,6 @@ void rpcSendResponse(const SRpcMsg *pRsp) {
if ( pConn->inType == 0 || pConn->user[0] == 0 ) {
tTrace("%s, connection is already released, rsp wont be sent", pConn->info);
rpcUnlockConn(pConn);
rpcDecRef(pRpc);
return;
}

6
src/vnode/src/vnodeRead.c
View File

@ -68,6 +68,7 @@ static void vnodeNotifyCurrentQhandle(void* handle, void* qhandle, int32_t vgId)
killQueryMsg->header.vgId = htonl(vgId);
killQueryMsg->header.contLen = htonl(sizeof(SRetrieveTableMsg));
vTrace("QInfo:%p register qhandle to connect:%p", qhandle, handle);
rpcReportProgress(handle, (char*) killQueryMsg, sizeof(SRetrieveTableMsg));
}
@ -85,10 +86,11 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
killQueryMsg->free = htons(killQueryMsg->free);
killQueryMsg->qhandle = htobe64(killQueryMsg->qhandle);
vWarn("QInfo:%p connection %p broken, kill query", killQueryMsg->qhandle, pReadMsg->rpcMsg.handle);
assert(pReadMsg->rpcMsg.contLen > 0 && killQueryMsg->free == 1);
qDestroyQueryInfo((qinfo_t) killQueryMsg->qhandle);
return TSDB_CODE_SUCCESS;
qKillQuery((qinfo_t) killQueryMsg->qhandle);
return TSDB_CODE_TSC_QUERY_CANCELLED; // todo change the error code
}
int32_t code = TSDB_CODE_SUCCESS;