From 853c22de287bbe02260503fba38877c862493b13 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 16 Nov 2022 10:55:19 +0800 Subject: [PATCH 01/17] fix: memory leak while create stream --- source/dnode/mnode/impl/src/mndScheduler.c | 1 + source/dnode/mnode/impl/src/mndStream.c | 3 +++ 2 files changed, 4 insertions(+) diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index 7e7c6ee0b6..e1886511b7 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -508,6 +508,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) { qDestroyQueryPlan(pPlan); return -1; } + sdbRelease(pSdb, pVgroup); } } qDestroyQueryPlan(pPlan); diff --git a/source/dnode/mnode/impl/src/mndStream.c b/source/dnode/mnode/impl/src/mndStream.c index 2cfb81a817..62247f2c2a 100644 --- a/source/dnode/mnode/impl/src/mndStream.c +++ b/source/dnode/mnode/impl/src/mndStream.c @@ -295,6 +295,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj, return -1; } pObj->sourceDbUid = pSourceDb->uid; + mndReleaseDb(pMnode, pSourceDb); memcpy(pObj->targetSTbName, pCreate->targetStbFullName, TSDB_TABLE_FNAME_LEN); @@ -307,6 +308,7 @@ static int32_t mndBuildStreamObjFromCreateReq(SMnode *pMnode, SStreamObj *pObj, pObj->targetStbUid = mndGenerateUid(pObj->targetSTbName, TSDB_TABLE_FNAME_LEN); pObj->targetDbUid = pTargetDb->uid; + mndReleaseDb(pMnode, pTargetDb); pObj->sql = pCreate->sql; pObj->ast = pCreate->ast; @@ -523,6 +525,7 @@ static int32_t mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStre tFreeSMCreateStbReq(&createReq); mndFreeStb(&stbObj); + mndReleaseDb(pMnode, pDb); return 0; _OVER: From 6c6d95d6cc2ff9d532da1c3de8b9022e60db4da7 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 16 Nov 2022 10:55:49 +0800 Subject: [PATCH 02/17] test: add asan case --- tests/parallel_test/cases.task | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index f0dff430b3..e45e82b277 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -119,7 +119,7 @@ ,,y,script,./test.sh -f tsim/parser/groupby-basic.sim ,,,script,./test.sh -f tsim/parser/groupby.sim ,,,script,./test.sh -f tsim/parser/having_child.sim -,,,script,./test.sh -f tsim/parser/having.sim +,,y,script,./test.sh -f tsim/parser/having.sim ,,y,script,./test.sh -f tsim/parser/import_commit1.sim ,,y,script,./test.sh -f tsim/parser/import_commit2.sim ,,y,script,./test.sh -f tsim/parser/import_commit3.sim @@ -220,7 +220,7 @@ ,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim ,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim ,,,script,./test.sh -f tsim/stream/distributeSession0.sim -,,,script,./test.sh -f tsim/stream/session0.sim +,,y,script,./test.sh -f tsim/stream/session0.sim ,,y,script,./test.sh -f tsim/stream/session1.sim ,,y,script,./test.sh -f tsim/stream/state0.sim ,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim From 3ada61c346f46257e9eb146cc12bd620c71f4e9d Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 16 Nov 2022 11:13:53 +0800 Subject: [PATCH 03/17] restore some invalid code modify --- source/libs/sync/inc/syncInt.h | 2 +- source/libs/sync/src/syncMain.c | 39 +++++++++++++++++++++++++++------ 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index e4d4b28b6b..0e0042d526 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -70,7 +70,7 @@ typedef struct SSyncTimer { uint64_t counter; int32_t timerMS; SRaftId destId; - SSyncHbTimerData hbData; + void* pData; } SSyncTimer; typedef struct SElectTimer { diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 514160235f..036127bf9c 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -665,12 +665,13 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { int32_t ret = 0; if (syncIsInit()) { - SSyncHbTimerData* pData = &pSyncTimer->hbData; + SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData)); pData->pSyncNode = pSyncNode; pData->pTimer = pSyncTimer; pData->destId = pSyncTimer->destId; pData->logicClock = pSyncTimer->logicClock; + pSyncTimer->pData = pData; taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, syncEnv()->pTimerManager, &pSyncTimer->pTimer); } else { sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId); @@ -1104,8 +1105,15 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { int32_t ret = 0; if (syncIsInit()) { pSyncNode->electTimerMS = ms; - taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager, + + SElectTimer* pElectTimer = taosMemoryMalloc(sizeof(SElectTimer)); + pElectTimer->logicClock = pSyncNode->electTimerLogicClock; + pElectTimer->pSyncNode = pSyncNode; + pElectTimer->pData = NULL; + + taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pElectTimer, syncEnv()->pTimerManager, &pSyncNode->pElectTimer); + } else { sError("vgId:%d, start elect timer error, sync env is stop", pSyncNode->vgId); } @@ -1855,28 +1863,45 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) { } static void syncNodeEqElectTimer(void* param, void* tmrId) { - SSyncNode* pNode = param; - if (!syncIsInit()) return; + + SElectTimer* pElectTimer = param; + SSyncNode* pNode = pElectTimer->pSyncNode; + if (pNode == NULL) return; if (pNode->syncEqMsg == NULL) return; SRpcMsg rpcMsg = {0}; - int32_t code = - syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerLogicClock, pNode->electTimerMS, pNode); + int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pElectTimer->logicClock, pNode->electTimerMS, pNode); + if (code != 0) { sError("failed to build elect msg"); + taosMemoryFree(pElectTimer); return; } SyncTimeout* pTimeout = rpcMsg.pCont; - sTrace("enqueue elect msg lc:%" PRId64, pTimeout->logicClock); + sNTrace(pNode, "enqueue elect msg lc:%" PRId64, pTimeout->logicClock); code = pNode->syncEqMsg(pNode->msgcb, &rpcMsg); if (code != 0) { sError("failed to sync enqueue elect msg since %s", terrstr()); rpcFreeCont(rpcMsg.pCont); + taosMemoryFree(pElectTimer); + return; } + + taosMemoryFree(pElectTimer); + +#if 0 + // reset timer ms + if (syncIsInit() && pNode->electBaseLine > 0) { + pNode->electTimerMS = syncUtilElectRandomMS(pNode->electBaseLine, 2 * pNode->electBaseLine); + taosTmrReset(syncNodeEqElectTimer, pNode->electTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pElectTimer); + } else { + sError("sync env is stop, syncNodeEqElectTimer"); + } +#endif } static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { From 7bcd13e82ae20e39d4b2217d79b8d7e8f9f94a87 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 16 Nov 2022 14:05:34 +0800 Subject: [PATCH 04/17] fix(sync): fix elect timer memory leak --- source/libs/sync/inc/syncInt.h | 14 +++++++---- source/libs/sync/src/syncMain.c | 44 ++++++++++++++------------------- source/libs/sync/src/syncUtil.c | 4 +-- 3 files changed, 29 insertions(+), 33 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 0e0042d526..0507bb3661 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -41,7 +41,7 @@ typedef struct SSyncRespMgr SSyncRespMgr; typedef struct SSyncSnapshotSender SSyncSnapshotSender; typedef struct SSyncSnapshotReceiver SSyncSnapshotReceiver; typedef struct SSyncTimer SSyncTimer; -typedef struct SSyncHbTimerData SSyncHbTimerData; +typedef struct SSyncHbTimerParam SSyncHbTimerParam; typedef struct SyncSnapshotSend SyncSnapshotSend; typedef struct SyncSnapshotRsp SyncSnapshotRsp; typedef struct SyncLocalCmd SyncLocalCmd; @@ -56,12 +56,13 @@ typedef struct SRaftId { SyncGroupId vgId; } SRaftId; -typedef struct SSyncHbTimerData { +typedef struct SSyncHbTimerParam { SSyncNode* pSyncNode; SSyncTimer* pTimer; SRaftId destId; uint64_t logicClock; -} SSyncHbTimerData; + int64_t executeTime; +} SSyncHbTimerParam; typedef struct SSyncTimer { void* pTimer; @@ -73,11 +74,12 @@ typedef struct SSyncTimer { void* pData; } SSyncTimer; -typedef struct SElectTimer { +typedef struct SElectTimerParam { uint64_t logicClock; SSyncNode* pSyncNode; + int64_t executeTime; void* pData; -} SElectTimer; +} SElectTimerParam; typedef struct SPeerState { SyncIndex lastSendIndex; @@ -153,6 +155,7 @@ typedef struct SSyncNode { uint64_t electTimerLogicClock; TAOS_TMR_CALLBACK FpElectTimerCB; // Timer Fp uint64_t electTimerCounter; + SElectTimerParam electTimerParam; // heartbeat timer tmr_h pHeartbeatTimer; @@ -161,6 +164,7 @@ typedef struct SSyncNode { uint64_t heartbeatTimerLogicClockUser; TAOS_TMR_CALLBACK FpHeartbeatTimerCB; // Timer Fp uint64_t heartbeatTimerCounter; + SSyncHbTimerParam hbTimerParam; // peer heartbeat timer SSyncTimer peerHeartbeatTimerArr[TSDB_MAX_REPLICA]; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 036127bf9c..7484e93abe 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -665,7 +665,7 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { int32_t ret = 0; if (syncIsInit()) { - SSyncHbTimerData* pData = taosMemoryMalloc(sizeof(SSyncHbTimerData)); + SSyncHbTimerParam* pData = taosMemoryMalloc(sizeof(SSyncHbTimerParam)); pData->pSyncNode = pSyncNode; pData->pTimer = pSyncTimer; pData->destId = pSyncTimer->destId; @@ -1106,12 +1106,13 @@ int32_t syncNodeStartElectTimer(SSyncNode* pSyncNode, int32_t ms) { if (syncIsInit()) { pSyncNode->electTimerMS = ms; - SElectTimer* pElectTimer = taosMemoryMalloc(sizeof(SElectTimer)); - pElectTimer->logicClock = pSyncNode->electTimerLogicClock; - pElectTimer->pSyncNode = pSyncNode; - pElectTimer->pData = NULL; + int64_t execTime = taosGetTimestampMs() + ms; + atomic_store_64(&(pSyncNode->electTimerParam.executeTime), execTime); + atomic_store_64(&(pSyncNode->electTimerParam.logicClock), pSyncNode->electTimerLogicClock); + pSyncNode->electTimerParam.pSyncNode = pSyncNode; + pSyncNode->electTimerParam.pData = NULL; - taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pElectTimer, syncEnv()->pTimerManager, + taosTmrReset(pSyncNode->FpElectTimerCB, pSyncNode->electTimerMS, pSyncNode, syncEnv()->pTimerManager, &pSyncNode->pElectTimer); } else { @@ -1865,18 +1866,21 @@ static void syncNodeEqPingTimer(void* param, void* tmrId) { static void syncNodeEqElectTimer(void* param, void* tmrId) { if (!syncIsInit()) return; - SElectTimer* pElectTimer = param; - SSyncNode* pNode = pElectTimer->pSyncNode; + SSyncNode* pNode = (SSyncNode*)param; if (pNode == NULL) return; if (pNode->syncEqMsg == NULL) return; + int64_t tsNow = taosGetTimestampMs(); + if (tsNow < pNode->electTimerParam.executeTime) return; + SRpcMsg rpcMsg = {0}; - int32_t code = syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pElectTimer->logicClock, pNode->electTimerMS, pNode); + int32_t code = + syncBuildTimeout(&rpcMsg, SYNC_TIMEOUT_ELECTION, pNode->electTimerParam.logicClock, pNode->electTimerMS, pNode); if (code != 0) { sError("failed to build elect msg"); - taosMemoryFree(pElectTimer); + return; } @@ -1887,21 +1891,9 @@ static void syncNodeEqElectTimer(void* param, void* tmrId) { if (code != 0) { sError("failed to sync enqueue elect msg since %s", terrstr()); rpcFreeCont(rpcMsg.pCont); - taosMemoryFree(pElectTimer); + return; } - - taosMemoryFree(pElectTimer); - -#if 0 - // reset timer ms - if (syncIsInit() && pNode->electBaseLine > 0) { - pNode->electTimerMS = syncUtilElectRandomMS(pNode->electBaseLine, 2 * pNode->electBaseLine); - taosTmrReset(syncNodeEqElectTimer, pNode->electTimerMS, pNode, syncEnv()->pTimerManager, &pNode->pElectTimer); - } else { - sError("sync env is stop, syncNodeEqElectTimer"); - } -#endif } static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { @@ -1938,9 +1930,9 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { } static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { - SSyncHbTimerData* pData = (SSyncHbTimerData*)param; - SSyncNode* pSyncNode = pData->pSyncNode; - SSyncTimer* pSyncTimer = pData->pTimer; + SSyncHbTimerParam* pData = (SSyncHbTimerParam*)param; + SSyncNode* pSyncNode = pData->pSyncNode; + SSyncTimer* pSyncTimer = pData->pTimer; if (pSyncNode == NULL) { return; diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index fb1b07b0b6..1d0ad07c99 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -193,7 +193,7 @@ static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) { } void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) { - if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; + if (pNode == NULL || pNode->pRaftCfg == NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; int64_t currentTerm = pNode->pRaftStore->currentTerm; // save error code, otherwise it will be overwritten @@ -554,4 +554,4 @@ void syncLogSendRequestVoteReply(SSyncNode* pSyncNode, const SyncRequestVoteRepl syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port); sNTrace(pSyncNode, "send sync-request-vote-reply to %s:%d {term:%" PRId64 ", grant:%d}, %s", host, port, pMsg->term, pMsg->voteGranted, s); -} \ No newline at end of file +} From 81592dfb09bc18b5ecc49e7efb21bf4c03fc32bb Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 16 Nov 2022 14:25:21 +0800 Subject: [PATCH 05/17] fix(sync): fix hb-timer memory leak --- source/libs/sync/inc/syncInt.h | 8 +++----- source/libs/sync/src/syncMain.c | 9 ++++----- 2 files changed, 7 insertions(+), 10 deletions(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index 0507bb3661..b7ca735ade 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -56,13 +56,12 @@ typedef struct SRaftId { SyncGroupId vgId; } SRaftId; -typedef struct SSyncHbTimerParam { +typedef struct SSyncHbTimerData { SSyncNode* pSyncNode; SSyncTimer* pTimer; SRaftId destId; uint64_t logicClock; - int64_t executeTime; -} SSyncHbTimerParam; +} SSyncHbTimerData; typedef struct SSyncTimer { void* pTimer; @@ -71,7 +70,7 @@ typedef struct SSyncTimer { uint64_t counter; int32_t timerMS; SRaftId destId; - void* pData; + SSyncHbTimerData hbData; } SSyncTimer; typedef struct SElectTimerParam { @@ -164,7 +163,6 @@ typedef struct SSyncNode { uint64_t heartbeatTimerLogicClockUser; TAOS_TMR_CALLBACK FpHeartbeatTimerCB; // Timer Fp uint64_t heartbeatTimerCounter; - SSyncHbTimerParam hbTimerParam; // peer heartbeat timer SSyncTimer peerHeartbeatTimerArr[TSDB_MAX_REPLICA]; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 7484e93abe..d47bf4f401 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -665,13 +665,12 @@ static int32_t syncHbTimerInit(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer, SRa static int32_t syncHbTimerStart(SSyncNode* pSyncNode, SSyncTimer* pSyncTimer) { int32_t ret = 0; if (syncIsInit()) { - SSyncHbTimerParam* pData = taosMemoryMalloc(sizeof(SSyncHbTimerParam)); + SSyncHbTimerData* pData = &pSyncTimer->hbData; pData->pSyncNode = pSyncNode; pData->pTimer = pSyncTimer; pData->destId = pSyncTimer->destId; pData->logicClock = pSyncTimer->logicClock; - pSyncTimer->pData = pData; taosTmrReset(pSyncTimer->timerCb, pSyncTimer->timerMS, pData, syncEnv()->pTimerManager, &pSyncTimer->pTimer); } else { sError("vgId:%d, start ctrl hb timer error, sync env is stop", pSyncNode->vgId); @@ -1930,9 +1929,9 @@ static void syncNodeEqHeartbeatTimer(void* param, void* tmrId) { } static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) { - SSyncHbTimerParam* pData = (SSyncHbTimerParam*)param; - SSyncNode* pSyncNode = pData->pSyncNode; - SSyncTimer* pSyncTimer = pData->pTimer; + SSyncHbTimerData* pData = (SSyncHbTimerData*)param; + SSyncNode* pSyncNode = pData->pSyncNode; + SSyncTimer* pSyncTimer = pData->pTimer; if (pSyncNode == NULL) { return; From 4c912aa95537baf46a7cdf4eb336e62652eff2ef Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 16 Nov 2022 14:28:20 +0800 Subject: [PATCH 06/17] fix(sync): fix hb-timer memory leak --- source/libs/sync/inc/syncInt.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/libs/sync/inc/syncInt.h b/source/libs/sync/inc/syncInt.h index b7ca735ade..362618fece 100644 --- a/source/libs/sync/inc/syncInt.h +++ b/source/libs/sync/inc/syncInt.h @@ -41,7 +41,7 @@ typedef struct SSyncRespMgr SSyncRespMgr; typedef struct SSyncSnapshotSender SSyncSnapshotSender; typedef struct SSyncSnapshotReceiver SSyncSnapshotReceiver; typedef struct SSyncTimer SSyncTimer; -typedef struct SSyncHbTimerParam SSyncHbTimerParam; +typedef struct SSyncHbTimerData SSyncHbTimerData; typedef struct SyncSnapshotSend SyncSnapshotSend; typedef struct SyncSnapshotRsp SyncSnapshotRsp; typedef struct SyncLocalCmd SyncLocalCmd; From 1114bdefb4db0fbc44deeae53a283fd49dbfae77 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 16 Nov 2022 14:31:17 +0800 Subject: [PATCH 07/17] fix: invalid log print --- source/libs/sync/src/syncUtil.c | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/source/libs/sync/src/syncUtil.c b/source/libs/sync/src/syncUtil.c index 1d0ad07c99..b50336cd63 100644 --- a/source/libs/sync/src/syncUtil.c +++ b/source/libs/sync/src/syncUtil.c @@ -193,7 +193,7 @@ static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) { } void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) { - if (pNode == NULL || pNode->pRaftCfg == NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; + if (pNode == NULL || pNode->pRaftCfg == NULL || pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; int64_t currentTerm = pNode->pRaftStore->currentTerm; // save error code, otherwise it will be overwritten @@ -252,7 +252,7 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotSender* pSender, const char* format, ...) { SSyncNode* pNode = pSender->pSyncNode; - if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; + if (pNode == NULL || pNode->pRaftCfg == NULL || pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0}; if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) { @@ -304,7 +304,7 @@ void syncPrintSnapshotSenderLog(const char* flags, ELogLevel level, int32_t dfla void syncPrintSnapshotReceiverLog(const char* flags, ELogLevel level, int32_t dflag, SSyncSnapshotReceiver* pReceiver, const char* format, ...) { SSyncNode* pNode = pReceiver->pSyncNode; - if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; + if (pNode == NULL || pNode->pRaftCfg == NULL || pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return; SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0}; if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) { From aae3475900656d189ecaca5e46f36ce14ce0dcc6 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Wed, 16 Nov 2022 14:36:51 +0800 Subject: [PATCH 08/17] fix(query): ASAN heap buffer overflow TD-20454 --- source/libs/function/src/builtinsimpl.c | 9 +++++++-- source/libs/scalar/src/sclfunc.c | 11 +++++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 10860d291f..ef04d2df65 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -4495,18 +4495,23 @@ bool histogramFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultIn pInfo->totalCount = 0; pInfo->normalized = 0; - int8_t binType = getHistogramBinType(varDataVal(pCtx->param[1].param.pz)); + char *binTypeStr = strndup(varDataVal(pCtx->param[1].param.pz), varDataLen(pCtx->param[1].param.pz)); + int8_t binType = getHistogramBinType(binTypeStr); + taosMemoryFree(binTypeStr); + if (binType == UNKNOWN_BIN) { return false; } - char* binDesc = varDataVal(pCtx->param[2].param.pz); + char* binDesc = strndup(varDataVal(pCtx->param[2].param.pz), varDataLen(pCtx->param[2].param.pz)); int64_t normalized = pCtx->param[3].param.i; if (normalized != 0 && normalized != 1) { return false; } if (!getHistogramBinDesc(pInfo, binDesc, binType, (bool)normalized)) { + taosMemoryFree(binDesc); return false; } + taosMemoryFree(binDesc); return true; } diff --git a/source/libs/scalar/src/sclfunc.c b/source/libs/scalar/src/sclfunc.c index 22fa99d75d..bb4a36a4ce 100644 --- a/source/libs/scalar/src/sclfunc.c +++ b/source/libs/scalar/src/sclfunc.c @@ -2748,14 +2748,19 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP int32_t numOfBins = 0; int32_t totalCount = 0; - int8_t binType = getHistogramBinType(varDataVal(pInput[1].columnData->pData)); - char *binDesc = varDataVal(pInput[2].columnData->pData); + char *binTypeStr = strndup(varDataVal(pInput[1].columnData->pData), varDataLen(pInput[1].columnData->pData)); + int8_t binType = getHistogramBinType(binTypeStr); + taosMemoryFree(binTypeStr); + + char *binDesc = strndup(varDataVal(pInput[2].columnData->pData), varDataLen(pInput[2].columnData->pData)); int64_t normalized = *(int64_t *)(pInput[3].columnData->pData); int32_t type = GET_PARAM_TYPE(pInput); if (!getHistogramBinDesc(&bins, &numOfBins, binDesc, binType, (bool)normalized)) { + taosMemoryFree(binDesc); return TSDB_CODE_FAILED; } + taosMemoryFree(binDesc); for (int32_t i = 0; i < pInput->numOfRows; ++i) { if (colDataIsNull_s(pInputData, i)) { @@ -2785,6 +2790,8 @@ int32_t histogramScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP } } + colInfoDataEnsureCapacity(pOutputData, numOfBins, false); + for (int32_t k = 0; k < numOfBins; ++k) { int32_t len; char buf[512] = {0}; From cf601b206fdad4b51a331a9257528f48ac8ea104 Mon Sep 17 00:00:00 2001 From: Ganlin Zhao Date: Wed, 16 Nov 2022 14:36:51 +0800 Subject: [PATCH 09/17] fix(query): ASAN heap buffer overflow TD-20454 --- source/libs/function/src/builtinsimpl.c | 17 +++++++++++++++++ source/libs/scalar/src/sclfunc.c | 18 +++++++++++++++++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index ef04d2df65..4e1feb75b1 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -4366,6 +4366,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t int32_t numOfParams = cJSON_GetArraySize(binDesc); int32_t startIndex; if (numOfParams != 4) { + cJSON_Delete(binDesc); return false; } @@ -4376,15 +4377,18 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t cJSON* infinity = cJSON_GetObjectItem(binDesc, "infinity"); if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { + cJSON_Delete(binDesc); return false; } if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 + cJSON_Delete(binDesc); return false; } if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { + cJSON_Delete(binDesc); return false; } @@ -4402,12 +4406,14 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t // linear bin process if (width->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble + i * width->valuedouble; if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; @@ -4416,22 +4422,26 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t // log bin process if (start->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; } } else { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } @@ -4446,6 +4456,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t } } else if (cJSON_IsArray(binDesc)) { /* user input bins */ if (binType != USER_INPUT_BIN) { + cJSON_Delete(binDesc); return false; } numOfBins = cJSON_GetArraySize(binDesc); @@ -4453,6 +4464,7 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t cJSON* bin = binDesc->child; if (bin == NULL) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } int i = 0; @@ -4460,16 +4472,19 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t intervals[i] = bin->valuedouble; if (!cJSON_IsNumber(bin)) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (i != 0 && intervals[i] <= intervals[i - 1]) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } bin = bin->next; i++; } } else { + cJSON_Delete(binDesc); return false; } @@ -4482,6 +4497,8 @@ static bool getHistogramBinDesc(SHistoFuncInfo* pInfo, char* binDescStr, int8_t } taosMemoryFree(intervals); + cJSON_Delete(binDesc); + return true; } diff --git a/source/libs/scalar/src/sclfunc.c b/source/libs/scalar/src/sclfunc.c index bb4a36a4ce..336bc6d533 100644 --- a/source/libs/scalar/src/sclfunc.c +++ b/source/libs/scalar/src/sclfunc.c @@ -2620,6 +2620,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin int32_t numOfParams = cJSON_GetArraySize(binDesc); int32_t startIndex; if (numOfParams != 4) { + cJSON_Delete(binDesc); return false; } @@ -2630,15 +2631,18 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin cJSON *infinity = cJSON_GetObjectItem(binDesc, "infinity"); if (!cJSON_IsNumber(start) || !cJSON_IsNumber(count) || !cJSON_IsBool(infinity)) { + cJSON_Delete(binDesc); return false; } if (count->valueint <= 0 || count->valueint > 1000) { // limit count to 1000 + cJSON_Delete(binDesc); return false; } if (isinf(start->valuedouble) || (width != NULL && isinf(width->valuedouble)) || (factor != NULL && isinf(factor->valuedouble)) || (count != NULL && isinf(count->valuedouble))) { + cJSON_Delete(binDesc); return false; } @@ -2656,12 +2660,14 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin // linear bin process if (width->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble + i * width->valuedouble; if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; @@ -2670,22 +2676,26 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin // log bin process if (start->valuedouble == 0) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (factor->valuedouble < 0 || factor->valuedouble == 0 || factor->valuedouble == 1) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } for (int i = 0; i < counter + 1; ++i) { intervals[startIndex] = start->valuedouble * pow(factor->valuedouble, i * 1.0); if (isinf(intervals[startIndex])) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } startIndex++; } } else { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } @@ -2700,6 +2710,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin } } else if (cJSON_IsArray(binDesc)) { /* user input bins */ if (binType != USER_INPUT_BIN) { + cJSON_Delete(binDesc); return false; } numOfBins = cJSON_GetArraySize(binDesc); @@ -2707,6 +2718,7 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin cJSON *bin = binDesc->child; if (bin == NULL) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } int i = 0; @@ -2714,16 +2726,19 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin intervals[i] = bin->valuedouble; if (!cJSON_IsNumber(bin)) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } if (i != 0 && intervals[i] <= intervals[i - 1]) { taosMemoryFree(intervals); + cJSON_Delete(binDesc); return false; } bin = bin->next; i++; } } else { + cJSON_Delete(binDesc); return false; } @@ -2735,8 +2750,9 @@ static bool getHistogramBinDesc(SHistoFuncBin **bins, int32_t *binNum, char *bin (*bins)[i].count = 0; } - cJSON_Delete(binDesc); taosMemoryFree(intervals); + cJSON_Delete(binDesc); + return true; } From ab57d069d4fe954fc1cad1e9ceb8ea8cf828b670 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Wed, 16 Nov 2022 15:18:31 +0800 Subject: [PATCH 10/17] enh(stream): add tbname map cache --- source/libs/executor/inc/executorimpl.h | 54 +++++++++++++------------ source/libs/executor/src/scanoperator.c | 32 ++++++++++++--- source/libs/wal/src/walWrite.c | 2 +- 3 files changed, 56 insertions(+), 32 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index b0da277cfb..0e2635459d 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -154,8 +154,8 @@ typedef struct { } SSchemaInfo; typedef struct { - int32_t operatorType; - int64_t refId; + int32_t operatorType; + int64_t refId; } SExchangeOpStopInfo; typedef struct { @@ -261,10 +261,10 @@ typedef struct SLimitInfo { } SLimitInfo; typedef struct SExchangeInfo { - SArray* pSources; - SArray* pSourceDataInfo; - tsem_t ready; - void* pTransporter; + SArray* pSources; + SArray* pSourceDataInfo; + tsem_t ready; + void* pTransporter; // SArray, result block list, used to keep the multi-block that // passed by downstream operator SArray* pResultBlockList; @@ -275,7 +275,7 @@ typedef struct SExchangeInfo { SLoadRemoteDataInfo loadInfo; uint64_t self; SLimitInfo limitInfo; - int64_t openedTs; // start exec time stamp + int64_t openedTs; // start exec time stamp } SExchangeInfo; typedef struct SScanInfo { @@ -310,9 +310,9 @@ typedef struct { } SAggOptrPushDownInfo; typedef struct STableMetaCacheInfo { - SLRUCache* pTableMetaEntryCache; // 100 by default - uint64_t metaFetch; - uint64_t cacheHit; + SLRUCache* pTableMetaEntryCache; // 100 by default + uint64_t metaFetch; + uint64_t cacheHit; } STableMetaCacheInfo; typedef struct STableScanInfo { @@ -343,7 +343,7 @@ typedef struct STableMergeScanInfo { int32_t tableEndIndex; bool hasGroupId; uint64_t groupId; - SArray* queryConds; // array of queryTableDataCond + SArray* queryConds; // array of queryTableDataCond STsdbReader* pReader; SReadHandle readHandle; int32_t bufPageSize; @@ -358,7 +358,7 @@ typedef struct STableMergeScanInfo { int64_t numOfRows; SScanInfo scanInfo; int32_t scanTimes; - SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context + SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context SResultRowInfo* pResultRowInfo; int32_t* rowEntryInfoOffset; SExprInfo* pExpr; @@ -504,6 +504,7 @@ typedef struct SStreamScanInfo { STimeWindow updateWin; STimeWindowAggSupp twAggSup; SSDataBlock* pUpdateDataRes; + SHashObj* pGroupIdTbNameMap; // status for tmq SNodeList* pGroupTags; SNode* pTagCond; @@ -550,10 +551,10 @@ typedef struct SSysTableScanInfo { } SSysTableScanInfo; typedef struct SBlockDistInfo { - SSDataBlock* pResBlock; - STsdbReader* pHandle; - SReadHandle readHandle; - uint64_t uid; // table uid + SSDataBlock* pResBlock; + STsdbReader* pHandle; + SReadHandle readHandle; + uint64_t uid; // table uid } SBlockDistInfo; // todo remove this @@ -663,9 +664,9 @@ typedef struct SGroupbyOperatorInfo { SAggSupporter aggSup; SArray* pGroupCols; // group by columns, SArray SArray* pGroupColVals; // current group column values, SArray - bool isInit; // denote if current val is initialized or not - char* keyBuf; // group by keys for hash - int32_t groupKeyLen; // total group by column width + bool isInit; // denote if current val is initialized or not + char* keyBuf; // group by keys for hash + int32_t groupKeyLen; // total group by column width SGroupResInfo groupResInfo; SExprSupp scalarSup; } SGroupbyOperatorInfo; @@ -890,14 +891,14 @@ STimeWindow getFirstQualifiedTimeWindow(int64_t ts, STimeWindow* pWindow, SInter int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag); int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz); -void doDestroyExchangeOperatorInfo(void* param); +void doDestroyExchangeOperatorInfo(void* param); void setOperatorCompleted(SOperatorInfo* pOperator); -void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status, void* pInfo, - SExecTaskInfo* pTaskInfo); +void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status, + void* pInfo, SExecTaskInfo* pTaskInfo); void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo); -int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, - SSDataBlock* pBlock, int32_t rows, const char* idStr, STableMetaCacheInfo * pCache); +int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock, + int32_t rows, const char* idStr, STableMetaCacheInfo* pCache); void cleanupAggSup(SAggSupporter* pAggSup); void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle); @@ -992,7 +993,7 @@ void setTaskKilled(SExecTaskInfo* pTaskInfo); void queryCostStatis(SExecTaskInfo* pTaskInfo); void doDestroyTask(SExecTaskInfo* pTaskInfo); -void destroyOperatorInfo(SOperatorInfo* pOperator); +void destroyOperatorInfo(SOperatorInfo* pOperator); int32_t getMaximumIdleDurationSec(); /* @@ -1038,6 +1039,7 @@ void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKE uint64_t* pGp, void* pTbName); void printDataBlock(SSDataBlock* pBlock, const char* flag); uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId); +void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock); int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); @@ -1061,7 +1063,7 @@ int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResul int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult); int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize); void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order); -int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo *pInfo); +int32_t qAppendTaskStopInfo(SExecTaskInfo* pTaskInfo, SExchangeOpStopInfo* pInfo); #ifdef __cplusplus } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 77d2aeba28..c93ca2b3f8 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1515,10 +1515,17 @@ static int32_t generateDeleteResultBlock(SStreamScanInfo* pInfo, SSDataBlock* pS for (int32_t i = 0; i < pSrcBlock->info.rows; i++) { uint64_t srcUid = srcUidData[i]; uint64_t groupId = srcGp[i]; + char* tbname[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN] = {0}; if (groupId == 0) { groupId = getGroupIdByData(pInfo, srcUid, srcStartTsCol[i], version); } - appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId, NULL); + if (pInfo->tbnameCalSup.pExprInfo) { + char* parTbname = taosHashGet(pInfo->pGroupIdTbNameMap, &groupId, sizeof(int64_t)); + memcpy(varDataVal(tbname), parTbname, TSDB_TABLE_NAME_LEN); + varDataSetLen(tbname, strlen(varDataVal(tbname))); + } + appendOneRowToStreamSpecialBlock(pDestBlock, srcStartTsCol + i, srcEndTsCol + i, srcUidData + i, &groupId, + tbname[0] == 0 ? NULL : tbname); } return TSDB_CODE_SUCCESS; } @@ -1562,9 +1569,16 @@ static void calBlockTag(SExprSupp* pTagCalSup, SSDataBlock* pBlock, SSDataBlock* blockDataDestroy(pSrcBlock); } -static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) { +void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) { + SExprSupp* pTbNameCalSup = &pInfo->tbnameCalSup; if (pTbNameCalSup == NULL || pTbNameCalSup->numOfExprs == 0) return; if (pBlock == NULL || pBlock->info.rows == 0) return; + if (pBlock->info.groupId) { + char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t)); + if (tbname != NULL) { + memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN); + } + } SSDataBlock* pSrcBlock = blockCopyOneRow(pBlock, 0); ASSERT(pSrcBlock->info.rows == 1); @@ -1592,6 +1606,11 @@ static void calBlockTbName(SExprSupp* pTbNameCalSup, SSDataBlock* pBlock) { pBlock->info.parTbName[0] = 0; } + if (pBlock->info.groupId) { + taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), pBlock->info.parTbName, + TSDB_TABLE_NAME_LEN); + } + blockDataDestroy(pSrcBlock); blockDataDestroy(pResBlock); } @@ -1713,7 +1732,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex); blockDataFreeRes((SSDataBlock*)pBlock); - calBlockTbName(&pInfo->tbnameCalSup, pInfo->pRes); + calBlockTbName(pInfo, pInfo->pRes); return 0; } @@ -1960,7 +1979,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) { SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp); if (pBlock != NULL) { - calBlockTbName(&pInfo->tbnameCalSup, pBlock); + calBlockTbName(pInfo, pBlock); updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); qDebug("stream recover scan get block, rows %d", pBlock->info.rows); return pBlock; @@ -2081,7 +2100,7 @@ FETCH_NEXT_BLOCK: pSDB->info.type = pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RANGE ? STREAM_NORMAL : STREAM_PULL_DATA; checkUpdateData(pInfo, true, pSDB, false); // printDataBlock(pSDB, "stream scan update"); - calBlockTbName(&pInfo->tbnameCalSup, pSDB); + calBlockTbName(pInfo, pSDB); return pSDB; } blockDataCleanup(pInfo->pUpdateDataRes); @@ -2386,6 +2405,7 @@ static void destroyStreamScanOperatorInfo(void* param) { } cleanupExprSupp(&pStreamScan->tbnameCalSup); + taosHashCleanup(pStreamScan->pGroupIdTbNameMap); updateInfoDestroy(pStreamScan->pUpdateInfo); blockDataDestroy(pStreamScan->pRes); @@ -2443,6 +2463,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys if (initExprSupp(&pInfo->tbnameCalSup, pSubTableExpr, 1) != 0) { goto _error; } + pInfo->pGroupIdTbNameMap = + taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK); } if (pTableScanNode->pTags != NULL) { diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 0bc76e4084..4360141e74 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -328,7 +328,7 @@ int32_t walEndSnapshot(SWal *pWal) { pInfo++; } if (POINTER_DISTANCE(pInfo, pWal->fileInfoSet->pData) > 0) { - wDebug("vgId:%d, wal end remove from %" PRId64, pWal->cfg.vgId, pInfo->firstVer); + wDebug("vgId:%d, wal end remove for %" PRId64, pWal->cfg.vgId, pInfo->firstVer); } else { wDebug("vgId:%d, wal no remove", pWal->cfg.vgId); } From 201ce10fbc1dd39492c6f159835dc478f34a1050 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 16 Nov 2022 15:45:40 +0800 Subject: [PATCH 11/17] fix: restore wal info changes --- source/dnode/mgmt/mgmt_vnode/src/vmWorker.c | 2 +- source/libs/sync/src/syncMain.c | 4 ++-- source/libs/wal/src/walWrite.c | 2 +- tests/parallel_test/cases.task | 12 ++++++------ 4 files changed, 10 insertions(+), 10 deletions(-) diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index 17bc1526b5..24240e82ef 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -287,7 +287,7 @@ int32_t vmGetQueueSize(SVnodeMgmt *pMgmt, int32_t vgId, EQueueType qtype) { vmReleaseVnode(pMgmt, pVnode); } if (size < 0) { - dError("vgId:%d, can't get size from queue since %s, qtype:%d", vgId, terrstr(), qtype); + dTrace("vgId:%d, can't get size from queue since %s, qtype:%d", vgId, terrstr(), qtype); size = 0; } return size; diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 514160235f..42bb500705 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -429,7 +429,7 @@ bool syncIsReadyForRead(int64_t rid) { int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) { if (pSyncNode->peersNum == 0) { - sDebug("only one replica, cannot leader transfer"); + sDebug("vgId:%d, only one replica, cannot leader transfer", pSyncNode->vgId); terrno = TSDB_CODE_SYN_ONE_REPLICA; return -1; } @@ -445,7 +445,7 @@ int32_t syncNodeLeaderTransfer(SSyncNode* pSyncNode) { int32_t syncNodeLeaderTransferTo(SSyncNode* pSyncNode, SNodeInfo newLeader) { if (pSyncNode->replicaNum == 1) { - sDebug("only one replica, cannot leader transfer"); + sDebug("vgId:%d, only one replica, cannot leader transfer", pSyncNode->vgId); terrno = TSDB_CODE_SYN_ONE_REPLICA; return -1; } diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 317ad129b0..9aa3dabf0c 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -325,7 +325,7 @@ int32_t walEndSnapshot(SWal *pWal) { SWalFileInfo *pInfo = taosArraySearch(pWal->fileInfoSet, &tmp, compareWalFileInfo, TD_LE); if (pInfo) { if (ver >= pInfo->lastVer) { - pInfo++; + pInfo--; } if (POINTER_DISTANCE(pInfo, pWal->fileInfoSet->pData) > 0) { wDebug("vgId:%d, wal end remove from %" PRId64, pWal->cfg.vgId, pInfo->firstVer); diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 0567f05d55..bb15cef204 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -37,10 +37,10 @@ ,,y,script,./test.sh -f tsim/db/taosdlog.sim ,,,script,./test.sh -f tsim/dnode/balance_replica1.sim ,,,script,./test.sh -f tsim/dnode/balance_replica3.sim -#,,,script,./test.sh -f tsim/dnode/balance1.sim -#,,,script,./test.sh -f tsim/dnode/balance2.sim -#,,,script,./test.sh -f tsim/dnode/balance3.sim -#,,,script,./test.sh -f tsim/dnode/balancex.sim +,,,script,./test.sh -f tsim/dnode/balance1.sim +,,,script,./test.sh -f tsim/dnode/balance2.sim +,,,script,./test.sh -f tsim/dnode/balance3.sim +,,,script,./test.sh -f tsim/dnode/balancex.sim ,,y,script,./test.sh -f tsim/dnode/create_dnode.sim ,,y,script,./test.sh -f tsim/dnode/drop_dnode_has_mnode.sim ,,y,script,./test.sh -f tsim/dnode/drop_dnode_has_qnode_snode.sim @@ -55,7 +55,7 @@ ,,,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v1_follower.sim ,,,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v2.sim ,,,script,./test.sh -f tsim/dnode/redistribute_vgroup_replica3_v3.sim -#,,,script,./test.sh -f tsim/dnode/vnode_clean.sim +,,,script,./test.sh -f tsim/dnode/vnode_clean.sim ,,,script,./test.sh -f tsim/dnode/use_dropped_dnode.sim ,,,script,./test.sh -f tsim/dnode/split_vgroup_replica1.sim ,,,script,./test.sh -f tsim/dnode/split_vgroup_replica3.sim @@ -296,7 +296,7 @@ ,,y,script,./test.sh -f tsim/vnode/replica3_vgroup.sim ,,y,script,./test.sh -f tsim/vnode/replica3_many.sim ,,y,script,./test.sh -f tsim/vnode/replica3_import.sim -#,,,script,./test.sh -f tsim/vnode/stable_balance_replica1.sim +,,,script,./test.sh -f tsim/vnode/stable_balance_replica1.sim ,,y,script,./test.sh -f tsim/vnode/stable_dnode2_stop.sim ,,y,script,./test.sh -f tsim/vnode/stable_dnode2.sim ,,y,script,./test.sh -f tsim/vnode/stable_dnode3.sim From 84c9b79b4dc55e4a952e3e47dc893ffa56043546 Mon Sep 17 00:00:00 2001 From: Minghao Li Date: Wed, 16 Nov 2022 16:12:35 +0800 Subject: [PATCH 12/17] fix(sync): fix hb-timer heap-use-after-free --- source/libs/sync/src/syncMain.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index 514160235f..af033eb7b3 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -2059,6 +2059,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) { SyncLocalCmd* pSyncMsg = rpcMsgLocalCmd.pCont; pSyncMsg->cmd = SYNC_LOCAL_CMD_FOLLOWER_CMT; pSyncMsg->fcIndex = pMsg->commitIndex; + SyncIndex fcIndex = pSyncMsg->fcIndex; if (ths->syncEqMsg != NULL && ths->msgcb != NULL) { int32_t code = ths->syncEqMsg(ths->msgcb, &rpcMsgLocalCmd); @@ -2066,7 +2067,7 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) { sError("vgId:%d, sync enqueue fc-commit msg error, code:%d", ths->vgId, code); rpcFreeCont(rpcMsgLocalCmd.pCont); } else { - sTrace("vgId:%d, sync enqueue fc-commit msg, fc-index:%" PRId64, ths->vgId, pSyncMsg->fcIndex); + sTrace("vgId:%d, sync enqueue fc-commit msg, fc-index:%" PRId64, ths->vgId, fcIndex); } } } From c928f03501a909c6625502722950b6b1a29f9387 Mon Sep 17 00:00:00 2001 From: WANG MINGMING Date: Wed, 16 Nov 2022 17:02:52 +0800 Subject: [PATCH 13/17] Update 13-schemaless.md --- docs/zh/14-reference/13-schemaless/13-schemaless.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/zh/14-reference/13-schemaless/13-schemaless.md b/docs/zh/14-reference/13-schemaless/13-schemaless.md index b89f625376..3aebd616a0 100644 --- a/docs/zh/14-reference/13-schemaless/13-schemaless.md +++ b/docs/zh/14-reference/13-schemaless/13-schemaless.md @@ -8,6 +8,8 @@ description: 'Schemaless 写入方式，可以免于预先创建超级表/子表 无模式写入方式建立的超级表及其对应的子表与通过 SQL 直接建立的超级表和子表完全没有区别，你也可以通过，SQL 语句直接向其中写入数据。需要注意的是，通过无模式写入方式建立的表，其表名是基于标签值按照固定的映射规则生成，所以无法明确地进行表意，缺乏可读性。 +注意：无模式写入会自动建表，不需要手动建表，手动建表的话可能会出现未知的错误。 + ## 无模式写入行协议 TDengine 的无模式写入的行协议兼容 InfluxDB 的 行协议（Line Protocol）、OpenTSDB 的 telnet 行协议、OpenTSDB 的 JSON 格式协议。但是使用这三种协议的时候，需要在 API 中指定输入内容使用解析协议的标准。 From 9dc37f2622fc22d43e9e5de530c2c4e7a9d83958 Mon Sep 17 00:00:00 2001 From: WANG MINGMING Date: Wed, 16 Nov 2022 17:05:20 +0800 Subject: [PATCH 14/17] Update 13-schemaless.md --- docs/en/14-reference/13-schemaless/13-schemaless.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/en/14-reference/13-schemaless/13-schemaless.md b/docs/en/14-reference/13-schemaless/13-schemaless.md index 5b7924ce56..10321ab083 100644 --- a/docs/en/14-reference/13-schemaless/13-schemaless.md +++ b/docs/en/14-reference/13-schemaless/13-schemaless.md @@ -8,6 +8,9 @@ will automatically add the required columns to ensure that the data written by t The schemaless writing method creates super tables and their corresponding subtables. These are completely indistinguishable from the super tables and subtables created directly via SQL. You can write data directly to them via SQL statements. Note that the names of tables created by schemaless writing are based on fixed mapping rules for tag values, so they are not explicitly ideographic and they lack readability. +Tips: +The schemaless write will automatically create a table. You do not need to create a table manually, or an unknown error may occur. + ## Schemaless Writing Line Protocol TDengine's schemaless writing line protocol supports InfluxDB's Line Protocol, OpenTSDB's telnet line protocol, and OpenTSDB's JSON format protocol. However, when using these three protocols, you need to specify in the API the standard of the parsing protocol to be used for the input content. From 636b3f6b30d86cb2529c2c6e00489f1fbe379910 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Wed, 16 Nov 2022 17:38:49 +0800 Subject: [PATCH 15/17] fix: add sc start/stop instruction to windows installer (#18211) * fix: sc create service in tdengine.iss * fix: upx on windows * fix: add sc start/stop to installer --- packaging/tools/windows_before_install.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packaging/tools/windows_before_install.txt b/packaging/tools/windows_before_install.txt index b793a3e801..ef783bf10b 100644 --- a/packaging/tools/windows_before_install.txt +++ b/packaging/tools/windows_before_install.txt @@ -1,3 +1,6 @@ TDengine is a high-efficient, scalable, high-available distributed time-series database, which makes a lot of optimizations on inserting and querying data, which is far more efficient than normal regular databases. So TDengine can meet the high requirements of IOT and other areas on storing and querying a large amount of data. -TDengine will be installed under C:\TDengine, users can modify configuration file C:\TDengine\cfg\taos.cfg, set the log file path or other parameters. \ No newline at end of file +TDengine will be installed under C:\TDengine, users can modify configuration file C:\TDengine\cfg\taos.cfg, set the log file path or other parameters. +To start/stop TDengine with administrator privileges: sc start/stop taosd +To start/stop taosAdapter with administrator privileges: sc start/stop taosadapter +Please manually remove C:\TDengine from your system PATH environment after you remove TDengine software. From 0154256507c45c2e6a60a3665c2098c2b32b8583 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Wed, 16 Nov 2022 18:13:26 +0800 Subject: [PATCH 16/17] test: add tsan case --- tests/parallel_test/cases.task | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 436f1882b0..4c9715270d 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -117,7 +117,7 @@ ,,y,script,./test.sh -f tsim/parser/fourArithmetic-basic.sim ,,y,script,./test.sh -f tsim/parser/function.sim ,,y,script,./test.sh -f tsim/parser/groupby-basic.sim -,,,script,./test.sh -f tsim/parser/groupby.sim +,,y,script,./test.sh -f tsim/parser/groupby.sim ,,y,script,./test.sh -f tsim/parser/having_child.sim ,,y,script,./test.sh -f tsim/parser/having.sim ,,y,script,./test.sh -f tsim/parser/import_commit1.sim From e2005cfa86a6c66603b3b1df80c10bd40277861c Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Wed, 16 Nov 2022 18:48:08 +0800 Subject: [PATCH 17/17] fix(stream): skip check update for scalar --- source/dnode/mnode/impl/src/mndSubscribe.c | 4 ++-- source/libs/executor/src/scanoperator.c | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index fd79846104..58c89d76aa 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -996,7 +996,7 @@ static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, numOfRows, (const char *)&pConsumerEp->consumerId, false); - mDebug("mnd show subscrptions: topic %s, consumer %" PRId64 "cgroup %s vgid %d", varDataVal(topic), + mDebug("mnd show subscriptions: topic %s, consumer %" PRId64 " cgroup %s vgid %d", varDataVal(topic), pConsumerEp->consumerId, varDataVal(cgroup), pVgEp->vgId); // offset @@ -1044,7 +1044,7 @@ static int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, numOfRows, NULL, true); - mDebug("mnd show subscrptions(unassigned): topic %s, cgroup %s vgid %d", varDataVal(topic), varDataVal(cgroup), + mDebug("mnd show subscriptions(unassigned): topic %s, cgroup %s vgid %d", varDataVal(topic), varDataVal(cgroup), pVgEp->vgId); // offset diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index c93ca2b3f8..229effc96b 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1980,7 +1980,9 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp); if (pBlock != NULL) { calBlockTbName(pInfo, pBlock); - updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); + if (pInfo->pUpdateInfo) { + updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); + } qDebug("stream recover scan get block, rows %d", pBlock->info.rows); return pBlock; }