From 59c6434a0d908ad529b7e44fe68b1bde4b75b5f8 Mon Sep 17 00:00:00 2001 From: hjxilinx Date: Fri, 24 Apr 2020 22:32:23 +0800 Subject: [PATCH] [td-187] 1. support group by query, 2. fix memory leaks in client side. 3. refactor mgmtConn --- src/client/inc/tscUtil.h | 4 ++-- src/client/inc/tsclient.h | 5 ++--- src/client/src/tscAsync.c | 2 +- src/client/src/tscSQLParser.c | 2 +- src/client/src/tscServer.c | 5 +++-- src/client/src/tscSql.c | 4 +++- src/client/src/tscStream.c | 6 +++--- src/client/src/tscSystem.c | 22 +++++++--------------- src/client/src/tscUtil.c | 28 ++++++++++++++++++++-------- tests/examples/c/demo.c | 4 ++-- 10 files changed, 44 insertions(+), 38 deletions(-) diff --git a/src/client/inc/tscUtil.h b/src/client/inc/tscUtil.h index c16945bbcf..d46c32d73d 100644 --- a/src/client/inc/tscUtil.h +++ b/src/client/inc/tscUtil.h @@ -196,14 +196,14 @@ STableMetaInfo* tscGetMetaInfo(SQueryInfo *pQueryInfo, int32_t tableIndex); SQueryInfo *tscGetQueryInfoDetail(SSqlCmd* pCmd, int32_t subClauseIndex); int32_t tscGetQueryInfoDetailSafely(SSqlCmd *pCmd, int32_t subClauseIndex, SQueryInfo** pQueryInfo); -void tscClearMeterMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache); +void tscClearTableMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache); STableMetaInfo* tscAddTableMetaInfo(SQueryInfo* pQueryInfo, const char* name, STableMeta* pTableMeta, SVgroupsInfo* vgroupList, SArray* pTagCols); STableMetaInfo* tscAddEmptyMetaInfo(SQueryInfo *pQueryInfo); int32_t tscAddSubqueryInfo(SSqlCmd *pCmd); -void tscFreeSubqueryInfo(SSqlCmd* pCmd); +void tscFreeQueryInfo(SSqlCmd* pCmd); void tscClearSubqueryInfo(SSqlCmd* pCmd); int tscGetSTableVgroupInfo(SSqlObj* pSql, int32_t clauseIndex); diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 9cc6bcb364..d6fb9279c7 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -298,6 +298,7 @@ typedef struct STscObj { char sversion[TSDB_VERSION_LEN]; char writeAuth : 1; char superAuth : 1; + void* pMgmtConn; struct SSqlObj * pSql; struct SSqlObj * pHb; struct SSqlObj * sqlList; @@ -359,7 +360,7 @@ typedef struct SSqlStream { struct SSqlStream *prev, *next; } SSqlStream; -int32_t tscInitRpc(const char *user, const char *secret); +int32_t tscInitRpc(const char *user, const char *secret, void** pMgmtConn); void tscInitMsgsFp(); int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion); @@ -427,9 +428,7 @@ int32_t tscToSQLCmd(SSqlObj *pSql, struct SSqlInfo *pInfo); char * tscGetResultColumnChr(SSqlRes *pRes, SQueryInfo *pQueryInfo, int32_t column); extern void * pVnodeConn; -extern void * pTscMgmtConn; extern void * tscCacheHandle; -extern int slaveIndex; extern void * tscTmr; extern void * tscQhandle; extern int tscKeepConn[]; diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index 99d20de48b..b954db0734 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -46,7 +46,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const pSql->signature = pSql; pSql->param = param; pSql->pTscObj = pObj; - pSql->maxRetry = 1; + pSql->maxRetry = TSDB_REPLICA_MAX_NUM; pSql->fp = fp; if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) { diff --git a/src/client/src/tscSQLParser.c b/src/client/src/tscSQLParser.c index 588e45c0b1..59e2f147e1 100644 --- a/src/client/src/tscSQLParser.c +++ b/src/client/src/tscSQLParser.c @@ -733,7 +733,7 @@ int32_t tscSetTableId(STableMetaInfo* pTableMetaInfo, SSQLToken* pzTableName, SS */ if (size > 0) { if (strncasecmp(oldName, pTableMetaInfo->name, tListLen(pTableMetaInfo->name)) != 0) { - tscClearMeterMetaInfo(pTableMetaInfo, false); + tscClearTableMetaInfo(pTableMetaInfo, false); } } else { assert(pTableMetaInfo->pTableMeta == NULL); diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 5ae8f2402b..c36102034b 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -188,6 +188,7 @@ void tscProcessActivityTimer(void *handle, void *tmrId) { } int tscSendMsgToServer(SSqlObj *pSql) { + STscObj* pObj = pSql->pTscObj; SSqlCmd* pCmd = &pSql->cmd; char *pMsg = rpcMallocCont(pCmd->payloadLen); @@ -221,7 +222,7 @@ int tscSendMsgToServer(SSqlObj *pSql) { .handle = pSql, .code = 0 }; - rpcSendRequest(pTscMgmtConn, &pSql->ipList, &rpcMsg); + rpcSendRequest(pObj->pMgmtConn, &pSql->ipList, &rpcMsg); } return TSDB_CODE_SUCCESS; @@ -2597,7 +2598,7 @@ int tscGetSTableVgroupInfo(SSqlObj *pSql, int32_t clauseIndex) { // if (pSql->fp != NULL && pSql->pStream == NULL) { // pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); -// tscFreeSubqueryInfo(pCmd); +// tscFreeQueryInfo(pCmd); // } tscTrace("%p allocate new pSqlObj:%p to get stable vgroupInfo", pSql, pNew); diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index f70beab166..4320e8f81b 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -66,7 +66,8 @@ STscObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con return NULL; } - if (tscInitRpc(user, pass) != 0) { + void* pMgmtConn = NULL; + if (tscInitRpc(user, pass, &pMgmtConn) != 0) { terrno = TSDB_CODE_NETWORK_UNAVAIL; return NULL; } @@ -118,6 +119,7 @@ STscObj *taosConnectImpl(const char *ip, const char *user, const char *pass, con strtolower(pObj->db, tmp); } + pObj->pMgmtConn = pMgmtConn; pthread_mutex_init(&pObj->mutex, NULL); SSqlObj *pSql = (SSqlObj *)calloc(1, sizeof(SSqlObj)); diff --git a/src/client/src/tscStream.c b/src/client/src/tscStream.c index 9f7d4887d1..d690681729 100644 --- a/src/client/src/tscStream.c +++ b/src/client/src/tscStream.c @@ -147,7 +147,7 @@ static void tscProcessStreamQueryCallback(void *param, TAOS_RES *tres, int numOf retryDelay); STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(&pStream->pSql->cmd, 0, 0); - tscClearMeterMetaInfo(pTableMetaInfo, true); + tscClearTableMetaInfo(pTableMetaInfo, true); tscSetRetryTimer(pStream, pStream->pSql, retryDelay); return; @@ -177,7 +177,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf if (pSql == NULL || numOfRows < 0) { int64_t retryDelayTime = tscGetRetryDelayTime(pStream->slidingTime, pStream->precision); tscError("%p stream:%p, retrieve data failed, code:%d, retry in %" PRId64 "ms", pSql, pStream, numOfRows, retryDelayTime); - tscClearMeterMetaInfo(pTableMetaInfo, true); + tscClearTableMetaInfo(pTableMetaInfo, true); tscSetRetryTimer(pStream, pStream->pSql, retryDelayTime); return; @@ -259,7 +259,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf pStream->numOfRes); // release the metric/meter meta information reference, so data in cache can be updated - tscClearMeterMetaInfo(pTableMetaInfo, false); + tscClearTableMetaInfo(pTableMetaInfo, false); tscSetNextLaunchTimer(pStream, pSql); } } diff --git a/src/client/src/tscSystem.c b/src/client/src/tscSystem.c index 6713f84f99..6966a58c45 100644 --- a/src/client/src/tscSystem.c +++ b/src/client/src/tscSystem.c @@ -33,9 +33,6 @@ // global, not configurable void * pVnodeConn; -void * pVMeterConn; -void * pTscMgmtConn; -void * pSlaveConn; void * tscCacheHandle; int slaveIndex; void * tscTmr; @@ -55,7 +52,7 @@ void tscCheckDiskUsage(void *para, void *unused) { taosTmrReset(tscCheckDiskUsage, 1000, NULL, tscTmr, &tscCheckDiskUsageTmr); } -int32_t tscInitRpc(const char *user, const char *secret) { +int32_t tscInitRpc(const char *user, const char *secret, void** pMgmtConn) { SRpcInit rpcInit; char secretEncrypt[32] = {0}; taosEncryptPass((uint8_t *)secret, strlen(secret), secretEncrypt); @@ -83,13 +80,13 @@ int32_t tscInitRpc(const char *user, const char *secret) { } // not stop service, switch users - if (strcmp(tsLastUser, user) != 0 && pTscMgmtConn != NULL) { + if (strcmp(tsLastUser, user) != 0 && *pMgmtConn != NULL) { tscTrace("switch user from %s to %s", user, tsLastUser); - rpcClose(pTscMgmtConn); - pTscMgmtConn = NULL; + rpcClose(*pMgmtConn); + *pMgmtConn = NULL; } - if (pTscMgmtConn == NULL) { + if (*pMgmtConn == NULL) { memset(&rpcInit, 0, sizeof(rpcInit)); rpcInit.localIp = tsLocalIp; rpcInit.localPort = 0; @@ -104,8 +101,8 @@ int32_t tscInitRpc(const char *user, const char *secret) { rpcInit.secret = secretEncrypt; strcpy(tsLastUser, user); - pTscMgmtConn = rpcOpen(&rpcInit); - if (pTscMgmtConn == NULL) { + *pMgmtConn = rpcOpen(&rpcInit); + if (*pMgmtConn == NULL) { tscError("failed to init connection to mgmt"); return -1; } @@ -219,11 +216,6 @@ void taos_cleanup() { pVnodeConn = NULL; } - if (pTscMgmtConn != NULL) { - rpcClose(pTscMgmtConn); - pTscMgmtConn = NULL; - } - taosTmrCleanUp(tscTmr); } diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index abdab2d9a1..e9ff558b92 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -337,7 +337,7 @@ void tscResetSqlCmdObj(SSqlCmd* pCmd) { pCmd->pTableList= NULL; pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); - tscFreeSubqueryInfo(pCmd); + tscFreeQueryInfo(pCmd); } /* @@ -761,6 +761,8 @@ void tscCloseTscObj(STscObj* pObj) { tscFreeSqlObj(pSql); sem_destroy(&pSql->rspSem); + rpcClose(pObj->pMgmtConn); + pthread_mutex_destroy(&pObj->mutex); tscTrace("%p DB connection is closed", pObj); @@ -1459,7 +1461,7 @@ bool tscShouldFreeHeatBeat(SSqlObj* pHb) { void tscCleanSqlCmd(SSqlCmd* pCmd) { pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); - tscFreeSubqueryInfo(pCmd); + tscFreeQueryInfo(pCmd); uint32_t allocSize = pCmd->allocSize; char* allocPtr = pCmd->payload; @@ -1601,7 +1603,7 @@ int32_t tscAddSubqueryInfo(SSqlCmd* pCmd) { return TSDB_CODE_SUCCESS; } -static void doClearSubqueryInfo(SQueryInfo* pQueryInfo) { +static void freeQueryInfoImpl(SQueryInfo* pQueryInfo) { tscTagCondRelease(&pQueryInfo->tagCond); tscFieldInfoClear(&pQueryInfo->fieldsInfo); @@ -1611,6 +1613,11 @@ static void doClearSubqueryInfo(SQueryInfo* pQueryInfo) { tscColumnListDestroy(pQueryInfo->colList); memset(&pQueryInfo->colList, 0, sizeof(pQueryInfo->colList)); + if (pQueryInfo->groupbyExpr.columnInfo != NULL) { + taosArrayDestroy(pQueryInfo->groupbyExpr.columnInfo); + pQueryInfo->groupbyExpr.columnInfo = NULL; + } + pQueryInfo->tsBuf = tsBufDestory(pQueryInfo->tsBuf); tfree(pQueryInfo->defaultVal); @@ -1619,11 +1626,11 @@ static void doClearSubqueryInfo(SQueryInfo* pQueryInfo) { void tscClearSubqueryInfo(SSqlCmd* pCmd) { for (int32_t i = 0; i < pCmd->numOfClause; ++i) { SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, i); - doClearSubqueryInfo(pQueryInfo); + freeQueryInfoImpl(pQueryInfo); } } -void tscFreeSubqueryInfo(SSqlCmd* pCmd) { +void tscFreeQueryInfo(SSqlCmd* pCmd) { if (pCmd == NULL || pCmd->numOfClause == 0) { return; } @@ -1632,7 +1639,7 @@ void tscFreeSubqueryInfo(SSqlCmd* pCmd) { char* addr = (char*)pCmd - offsetof(SSqlObj, cmd); SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, i); - doClearSubqueryInfo(pQueryInfo); + freeQueryInfoImpl(pQueryInfo); tscClearAllTableMetaInfo(pQueryInfo, (const char*)addr, false); tfree(pQueryInfo); } @@ -1691,7 +1698,7 @@ void doRemoveTableMetaInfo(SQueryInfo* pQueryInfo, int32_t index, bool removeFro STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, index); - tscClearMeterMetaInfo(pTableMetaInfo, removeFromCache); + tscClearTableMetaInfo(pTableMetaInfo, removeFromCache); free(pTableMetaInfo); int32_t after = pQueryInfo->numOfTables - index - 1; @@ -1713,13 +1720,18 @@ void tscClearAllTableMetaInfo(SQueryInfo* pQueryInfo, const char* address, bool tfree(pQueryInfo->pTableMetaInfo); } -void tscClearMeterMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache) { +void tscClearTableMetaInfo(STableMetaInfo* pTableMetaInfo, bool removeFromCache) { if (pTableMetaInfo == NULL) { return; } taosCacheRelease(tscCacheHandle, (void**)&(pTableMetaInfo->pTableMeta), removeFromCache); tfree(pTableMetaInfo->vgroupList); + + if (pTableMetaInfo->tagColList != NULL) { + taosArrayDestroy(pTableMetaInfo->tagColList); + pTableMetaInfo->tagColList = NULL; + } } void tscResetForNextRetrieve(SSqlRes* pRes) { diff --git a/tests/examples/c/demo.c b/tests/examples/c/demo.c index 9847300024..e5a70740f1 100644 --- a/tests/examples/c/demo.c +++ b/tests/examples/c/demo.c @@ -75,7 +75,7 @@ int main(int argc, char *argv[]) { doQuery(taos, "create database if not exists test"); doQuery(taos, "use test"); - doQuery(taos, "insert into tm99 values('2020-01-01 1:1:1', 99);"); + doQuery(taos, "select count(*),k,sum(k) from m1 group by k"); // doQuery(taos, "create table if not exists tm0 (ts timestamp, k int);"); // doQuery(taos, "insert into tm0 values('2020-1-1 1:1:1', 1);"); // doQuery(taos, "insert into tm0 values('2020-1-1 1:1:2', 2);"); @@ -86,7 +86,7 @@ int main(int argc, char *argv[]) { // doQuery(taos, "insert into tm0 values('2020-1-1 1:1:7', 7);"); // doQuery(taos, "insert into tm0 values('2020-1-1 1:1:8', 8);"); // doQuery(taos, "insert into tm0 values('2020-1-1 1:1:9', 9);"); - doQuery(taos, "select sum(k),count(*) from m1 group by a"); +// doQuery(taos, "select sum(k),count(*) from m1 group by a"); taos_close(taos); return 0; -- GitLab