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

[td-187] 1. support group by query, 2. fix memory leaks in client side. 3. refactor mgmtConn

This commit is contained in:
hjxilinx 2020-04-24 22:32:23 +08:00
parent 7ddbf61980
commit 59c6434a0d
10 changed files with 44 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/client/inc/tscUtil.h
View File

@ -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);

5
src/client/inc/tsclient.h
View File

@ -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[];

2
src/client/src/tscAsync.c
View File

@ -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)) {

2
src/client/src/tscSQLParser.c
View File

@ -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);

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

@ -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);

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

@ -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));

6
src/client/src/tscStream.c
View File

@ -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);
}
}

22
src/client/src/tscSystem.c
View File

@ -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);
}

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

@ -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) {

4
tests/examples/c/demo.c
View File

@ -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;