diff --git a/src/client/src/tscParseLineProtocol.c b/src/client/src/tscParseLineProtocol.c index c6cdee6f1f479da4b71163c9b507d4ae8889b4bb..7d2823a42edc3ea7616e822d47cfdbf8706f4a56 100644 --- a/src/client/src/tscParseLineProtocol.c +++ b/src/client/src/tscParseLineProtocol.c @@ -484,7 +484,7 @@ int32_t loadTableMeta(TAOS* taos, char* tableName, SSmlSTableSchema* schema, SSm size_t tagIndex = taosArrayGetSize(schema->tags) - 1; taosHashPut(schema->tagHash, field.name, strlen(field.name), &tagIndex, sizeof(tagIndex)); } - tscDebug("SML:0x%"PRIx64 "load table meta succeed. %s, columns number: %d, tag number: %d, precision: %d", + tscDebug("SML:0x%"PRIx64 " load table meta succeed. table name: %s, columns number: %d, tag number: %d, precision: %d", info->id, tableName, tableMeta->tableInfo.numOfColumns, tableMeta->tableInfo.numOfTags, schema->precision); free(tableMeta); tableMeta = NULL; return code; diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 7be3bbb3aadd440d8da63a6dc7297a5360fbd76e..fdb1be9f4ecdc0be2c313831c3d9c2f70dbfca96 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -337,11 +337,16 @@ int tscSendMsgToServer(SSqlObj *pSql) { return TSDB_CODE_SUCCESS; } -void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { +static void doProcessMsgFromServer(SSchedMsg* pSchedMsg) { + SRpcMsg* rpcMsg = pSchedMsg->ahandle; + SRpcEpSet* pEpSet = pSchedMsg->thandle; + TSDB_CACHE_PTR_TYPE handle = (TSDB_CACHE_PTR_TYPE) rpcMsg->ahandle; SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, handle); if (pSql == NULL) { rpcFreeCont(rpcMsg->pCont); + free(rpcMsg); + free(pEpSet); return; } @@ -359,6 +364,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { taosRemoveRef(tscObjRef, handle); taosReleaseRef(tscObjRef, handle); rpcFreeCont(rpcMsg->pCont); + free(rpcMsg); + free(pEpSet); return; } @@ -370,6 +377,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { taosRemoveRef(tscObjRef, handle); taosReleaseRef(tscObjRef, handle); rpcFreeCont(rpcMsg->pCont); + free(rpcMsg); + free(pEpSet); return; } @@ -425,6 +434,8 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { if (rpcMsg->code == TSDB_CODE_TSC_ACTION_IN_PROGRESS) { taosReleaseRef(tscObjRef, handle); rpcFreeCont(rpcMsg->pCont); + free(rpcMsg); + free(pEpSet); return; } } @@ -432,7 +443,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { } pRes->rspLen = 0; - + if (pRes->code == TSDB_CODE_TSC_QUERY_CANCELLED) { tscDebug("0x%"PRIx64" query is cancelled, code:%s", pSql->self, tstrerror(pRes->code)); } else { @@ -481,7 +492,7 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { tscDebug("0x%"PRIx64" SQL cmd:%s, code:%s rspLen:%d", pSql->self, sqlCmd[pCmd->command], tstrerror(pRes->code), pRes->rspLen); } } - + if (pRes->code == TSDB_CODE_SUCCESS && tscProcessMsgRsp[pCmd->command]) { rpcMsg->code = (*tscProcessMsgRsp[pCmd->command])(pSql); } @@ -502,6 +513,29 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { taosReleaseRef(tscObjRef, handle); rpcFreeCont(rpcMsg->pCont); + free(rpcMsg); + free(pEpSet); +} + +void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) { + SSchedMsg schedMsg = {0}; + + schedMsg.fp = doProcessMsgFromServer; + + SRpcMsg* rpcMsgCopy = calloc(1, sizeof(SRpcMsg)); + memcpy(rpcMsgCopy, rpcMsg, sizeof(struct SRpcMsg)); + schedMsg.ahandle = (void*)rpcMsgCopy; + + SRpcEpSet* pEpSetCopy = NULL; + if (pEpSet != NULL) { + pEpSetCopy = calloc(1, sizeof(SRpcEpSet)); + memcpy(pEpSetCopy, pEpSet, sizeof(SRpcEpSet)); + } + + schedMsg.thandle = (void*)pEpSetCopy; + schedMsg.msg = NULL; + + taosScheduleTask(tscQhandle, &schedMsg); } int doBuildAndSendMsg(SSqlObj *pSql) { diff --git a/src/query/src/qAggMain.c b/src/query/src/qAggMain.c index d96b260b13cf8a1bbcbc9b329a51bc1a714aba8d..1ed63be07070f4cad357025a92b5e2413af3f639 100644 --- a/src/query/src/qAggMain.c +++ b/src/query/src/qAggMain.c @@ -4132,6 +4132,11 @@ void blockinfo_func_finalizer(SQLFunctionCtx* pCtx) { pDist->rowSize = (uint16_t)pCtx->param[0].i64; generateBlockDistResult(pDist, pCtx->pOutput); + if (pDist->dataBlockInfos != NULL) { + taosArrayDestroy(pDist->dataBlockInfos); + pDist->dataBlockInfos = NULL; + } + // cannot set the numOfIteratedElems again since it is set during previous iteration pResInfo->numOfRes = 1; pResInfo->hasResult = DATA_SET_FLAG; diff --git a/tests/pytest/insert/metadataUpdate.py b/tests/pytest/insert/metadataUpdate.py index 1a960a20e60d6bbb25b08a47725b262e1b49da49..f996a707ffc8a98a49f4dc607cb19316d3f1085a 100644 --- a/tests/pytest/insert/metadataUpdate.py +++ b/tests/pytest/insert/metadataUpdate.py @@ -16,7 +16,6 @@ from util.log import tdLog from util.cases import tdCases from util.sql import tdSql from util.dnodes import tdDnodes -from multiprocessing import Process import subprocess class TDTestCase: @@ -28,16 +27,6 @@ class TDTestCase: self.tables = 10 self.rows = 1000 - def updateMetadata(self): - self.host = "127.0.0.1" - self.user = "root" - self.password = "taosdata" - self.config = tdDnodes.getSimCfgPath() - - self.conn = taos.connect(host = self.host, user = self.user, password = self.password, config = self.config) - self.cursor = self.conn.cursor() - self.cursor.execute("alter table db.tb add column col2 int") - print("alter table done") def deleteTableAndRecreate(self): self.config = tdDnodes.getSimCfgPath() @@ -68,11 +57,15 @@ class TDTestCase: tdSql.query("select * from tb") tdSql.checkRows(1) - p = Process(target=self.updateMetadata, args=()) - p.start() - p.join() - p.terminate() - + self.config = tdDnodes.getSimCfgPath() + command = ["taos", "-c", self.config, "-s", "alter table db.tb add column col2 int;"] + print("alter table db.tb add column col2 int;") + result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8") + if result.returncode == 0: + print("success:", result) + else: + print("error:", result) + tdSql.execute("insert into tb(ts, col1, col2) values(%d, 1, 2)" % (self.ts + 2)) print("==============step2")