#5056

Committed 17 May 2026 01:15AM UTC coverage: 73.384% (+0.03%) from 73.355%

Build # #5056

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

Coverage Stats

281643 of 383795 relevant lines covered (73.38%)

135942701.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.98

/source/dnode/mnode/impl/src/mndQuery.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include "mndQuery.h"
#include "executor.h"
#include "mndMnode.h"
#include "qworker.h"

int32_t mndPreProcessQueryMsg(SRpcMsg *pMsg, int32_t* qType) {
  SMnode *pMnode = pMsg->info.node;
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {
    return 0;
  }

  return qWorkerPreprocessQueryMsg(pMnode->pQuery, pMsg, false, qType);
}

void mndPostProcessQueryMsg(SRpcMsg *pMsg) {
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) return;
  SMnode *pMnode = pMsg->info.node;
  (void)qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg);
}

int32_t mndProcessQueryMsg(SRpcMsg *pMsg, SQueueInfo *pInfo) {
  int32_t code = -1;
  SMnode *pMnode = pMsg->info.node;

  SReadHandle handle = {0};
  handle =  (SReadHandle){.mnd = pMnode, .pMsgCb = &pMnode->msgCb, .pWorkerCb = pInfo->workerCb};

  mTrace("msg:%p, in query queue is processing", pMsg);
  switch (pMsg->msgType) {
    case TDMT_SCH_QUERY:
    case TDMT_SCH_MERGE_QUERY:
      code = qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pMsg, 0);
      break;
    case TDMT_SCH_QUERY_CONTINUE:
      code = qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pMsg, 0);
      break;
    case TDMT_SCH_FETCH:
    case TDMT_SCH_MERGE_FETCH:
      code = qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg, 0);
      break;
    case TDMT_SCH_DROP_TASK:
      code = qWorkerProcessDropMsg(pMnode, pMnode->pQuery, pMsg, 0);
      break;
    case TDMT_SCH_QUERY_HEARTBEAT:
      code = qWorkerProcessHbMsg(pMnode, pMnode->pQuery, pMsg, 0);
      break;
    case TDMT_SCH_TASK_NOTIFY:
      code = qWorkerProcessNotifyMsg(pMnode, pMnode->pQuery, pMsg, 0);
      break;
    default:
      terrno = TSDB_CODE_APP_ERROR;
      mError("unknown msg type:%d in query queue", pMsg->msgType);
  }

  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
  return code;
}

static FORCE_INLINE void mnodeFreeSBatchRspMsg(void *p) {
  if (NULL == p) {
    return;
  }

  SBatchRspMsg *pRsp = (SBatchRspMsg *)p;
  rpcFreeCont(pRsp->msg);
}

int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
  int32_t      code = 0;
  int32_t      rspSize = 0;
  SBatchReq    batchReq = {0};
  SBatchMsg    req = {0};
  SBatchRspMsg rsp = {0};
  SBatchRsp    batchRsp = {0};
  SRpcMsg      reqMsg = *pMsg;
  void        *pRsp = NULL;
  SMnode      *pMnode = pMsg->info.node;

  if ((code = tDeserializeSBatchReq(pMsg->pCont, pMsg->contLen, &batchReq)) != 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    mError("tDeserializeSBatchReq failed");
    goto _exit;
  }

  int32_t msgNum = taosArrayGetSize(batchReq.pMsgs);
  if (msgNum >= MAX_META_MSG_IN_BATCH) {
    code = TSDB_CODE_INVALID_MSG;
    mError("too many msgs %d in mnode batch meta req", msgNum);
    goto _exit;
  }

  batchRsp.pRsps = taosArrayInit(msgNum, sizeof(SBatchRspMsg));
  if (NULL == batchRsp.pRsps) {
    code = terrno;
    goto _exit;
  }

  for (int32_t i = 0; i < msgNum; ++i) {
    SBatchMsg *req = taosArrayGet(batchReq.pMsgs, i);

    reqMsg.msgType = req->msgType;
    reqMsg.pCont = req->msg;
    reqMsg.contLen = req->msgLen;
    reqMsg.info.rsp = NULL;
    reqMsg.info.rspLen = 0;

    MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(req->msgType)];
    if (fp == NULL) {
      mError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));
      code = TSDB_CODE_MSG_NOT_PROCESSED;
      taosArrayDestroy(batchRsp.pRsps);
      return -1;
    }

    if ((*fp)(&reqMsg)) {
      rsp.rspCode = terrno;
    } else {
      rsp.rspCode = 0;
    }
    rsp.msgIdx = req->msgIdx;
    rsp.reqType = reqMsg.msgType;
    rsp.msgLen = reqMsg.info.rspLen;
    rsp.msg = reqMsg.info.rsp;

    if (taosArrayPush(batchRsp.pRsps, &rsp) == NULL) {
      mError("msg:%p, failed to put array since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle,
             TMSG_INFO(pMsg->msgType));
    }
  }

  rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);
  if (rspSize < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  pRsp = rpcMallocCont(rspSize);
  if (pRsp == NULL) {
    code = terrno;
    goto _exit;
  }
  if (tSerializeSBatchRsp(pRsp, rspSize, &batchRsp) < 0) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

_exit:

  pMsg->info.rsp = pRsp;
  pMsg->info.rspLen = rspSize;

  if (code) {
    mError("mnd get batch meta failed cause of %s", tstrerror(code));
  }

  taosArrayDestroyEx(batchReq.pMsgs, tFreeSBatchReqMsg);
  taosArrayDestroyEx(batchRsp.pRsps, mnodeFreeSBatchRspMsg);

  TAOS_RETURN(code);
}

int32_t mndInitQuery(SMnode *pMnode) {
  if (qWorkerInit(NODE_TYPE_MNODE, MNODE_HANDLE, (void **)&pMnode->pQuery, &pMnode->msgCb) != 0) {
    mError("failed to init qworker in mnode since %s", terrstr());
    return -1;
  }

  mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_MERGE_QUERY, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY_CONTINUE, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_FETCH, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_MERGE_FETCH, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_TASK_NOTIFY, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_DROP_TASK, mndProcessQueryMsg);
  mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY_HEARTBEAT, mndProcessQueryMsg);
  mndSetMsgHandle(pMnode, TDMT_MND_BATCH_META, mndProcessBatchMetaMsg);

  return 0;
}

void mndCleanupQuery(SMnode *pMnode) { qWorkerDestroy((void **)&pMnode->pQuery); }

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "mndQuery.h"
17	#include "executor.h"
18	#include "mndMnode.h"
19	#include "qworker.h"
20
21	int32_t mndPreProcessQueryMsg(SRpcMsg pMsg, int32_t qType) {	6,650,919✔
22	SMnode *pMnode = pMsg->info.node;	6,650,919✔
23	if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {	6,650,919✔
24	return 0;	×
25	}
26
27	return qWorkerPreprocessQueryMsg(pMnode->pQuery, pMsg, false, qType);	6,650,919✔
28	}
29
30	void mndPostProcessQueryMsg(SRpcMsg *pMsg) {	6,179,166✔
31	if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) return;	6,179,166✔
32	SMnode *pMnode = pMsg->info.node;	×
33	(void)qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg);	×
34	}
35
36	int32_t mndProcessQueryMsg(SRpcMsg pMsg, SQueueInfo pInfo) {	25,426,377✔
37	int32_t code = -1;	25,426,377✔
38	SMnode *pMnode = pMsg->info.node;	25,426,377✔
39
40	SReadHandle handle = {0};	25,426,377✔
41	handle = (SReadHandle){.mnd = pMnode, .pMsgCb = &pMnode->msgCb, .pWorkerCb = pInfo->workerCb};	25,426,377✔
42
43	mTrace("msg:%p, in query queue is processing", pMsg);	25,423,576✔
44	switch (pMsg->msgType) {	25,423,576✔
45	case TDMT_SCH_QUERY:	6,650,919✔
46	case TDMT_SCH_MERGE_QUERY:
47	code = qWorkerProcessQueryMsg(&handle, pMnode->pQuery, pMsg, 0);	6,650,919✔
48	break;	6,650,919✔
49	case TDMT_SCH_QUERY_CONTINUE:	382✔
50	code = qWorkerProcessCQueryMsg(&handle, pMnode->pQuery, pMsg, 0);	382✔
51	break;	382✔
52	case TDMT_SCH_FETCH:	6,490,769✔
53	case TDMT_SCH_MERGE_FETCH:
54	code = qWorkerProcessFetchMsg(pMnode, pMnode->pQuery, pMsg, 0);	6,490,769✔
55	break;	6,490,769✔
56	case TDMT_SCH_DROP_TASK:	6,627,145✔
57	code = qWorkerProcessDropMsg(pMnode, pMnode->pQuery, pMsg, 0);	6,627,145✔
58	break;	6,627,145✔
59	case TDMT_SCH_QUERY_HEARTBEAT:	5,657,365✔
60	code = qWorkerProcessHbMsg(pMnode, pMnode->pQuery, pMsg, 0);	5,657,365✔
61	break;	5,657,365✔
62	case TDMT_SCH_TASK_NOTIFY:	×
63	code = qWorkerProcessNotifyMsg(pMnode, pMnode->pQuery, pMsg, 0);	×
64	break;	×
65	default:	×
66	terrno = TSDB_CODE_APP_ERROR;	×
67	mError("unknown msg type:%d in query queue", pMsg->msgType);	×
68	}
69
70	if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;	25,426,580✔
71	return code;	25,426,580✔
72	}
73
74	static FORCE_INLINE void mnodeFreeSBatchRspMsg(void *p) {	73,274,453✔
75	if (NULL == p) {	73,274,453✔
76	return;	×
77	}
78
79	SBatchRspMsg pRsp = (SBatchRspMsg )p;	73,274,453✔
80	rpcFreeCont(pRsp->msg);	73,274,453✔
81	}
82
83	int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {	37,012,211✔
84	int32_t code = 0;	37,012,211✔
85	int32_t rspSize = 0;	37,012,211✔
86	SBatchReq batchReq = {0};	37,012,211✔
87	SBatchMsg req = {0};	37,012,211✔
88	SBatchRspMsg rsp = {0};	37,012,211✔
89	SBatchRsp batchRsp = {0};	37,011,047✔
90	SRpcMsg reqMsg = *pMsg;	37,010,989✔
91	void *pRsp = NULL;	37,010,989✔
92	SMnode *pMnode = pMsg->info.node;	37,010,989✔
93
94	if ((code = tDeserializeSBatchReq(pMsg->pCont, pMsg->contLen, &batchReq)) != 0) {	37,013,807✔
95	code = TSDB_CODE_OUT_OF_MEMORY;	×
96	mError("tDeserializeSBatchReq failed");	×
97	goto _exit;	×
98	}
99
100	int32_t msgNum = taosArrayGetSize(batchReq.pMsgs);	37,010,615✔
101	if (msgNum >= MAX_META_MSG_IN_BATCH) {	37,011,426✔
102	code = TSDB_CODE_INVALID_MSG;	×
103	mError("too many msgs %d in mnode batch meta req", msgNum);	×
104	goto _exit;	×
105	}
106
107	batchRsp.pRsps = taosArrayInit(msgNum, sizeof(SBatchRspMsg));	37,011,426✔
108	if (NULL == batchRsp.pRsps) {	37,009,375✔
109	code = terrno;	×
110	goto _exit;	×
111	}
112
113	for (int32_t i = 0; i < msgNum; ++i) {	110,285,359✔
114	SBatchMsg *req = taosArrayGet(batchReq.pMsgs, i);	73,273,181✔
115
116	reqMsg.msgType = req->msgType;	73,270,678✔
117	reqMsg.pCont = req->msg;	73,271,531✔
118	reqMsg.contLen = req->msgLen;	73,272,387✔
119	reqMsg.info.rsp = NULL;	73,271,164✔
120	reqMsg.info.rspLen = 0;	73,271,164✔
121
122	MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(req->msgType)];	73,271,164✔
123	if (fp == NULL) {	73,271,047✔
124	mError("msg:%p, failed to get msg handle, app:%p type:%s", pMsg, pMsg->info.ahandle, TMSG_INFO(pMsg->msgType));	×
125	code = TSDB_CODE_MSG_NOT_PROCESSED;	×
126	taosArrayDestroy(batchRsp.pRsps);	×
127	return -1;	×
128	}
129
130	if ((*fp)(&reqMsg)) {	73,271,047✔
131	rsp.rspCode = terrno;	21,660,899✔
132	} else {
133	rsp.rspCode = 0;	51,613,875✔
134	}
135	rsp.msgIdx = req->msgIdx;	73,273,951✔
136	rsp.reqType = reqMsg.msgType;	73,273,938✔
137	rsp.msgLen = reqMsg.info.rspLen;	73,273,938✔
138	rsp.msg = reqMsg.info.rsp;	73,273,938✔
139
140	if (taosArrayPush(batchRsp.pRsps, &rsp) == NULL) {	146,548,712✔
141	mError("msg:%p, failed to put array since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle,	×
142	TMSG_INFO(pMsg->msgType));
143	}
144	}
145
146	rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);	37,012,178✔
147	if (rspSize < 0) {	37,013,590✔
148	code = TSDB_CODE_OUT_OF_MEMORY;	×
149	goto _exit;	×
150	}
151	pRsp = rpcMallocCont(rspSize);	37,013,590✔
152	if (pRsp == NULL) {	37,013,536✔
153	code = terrno;	×
154	goto _exit;	×
155	}
156	if (tSerializeSBatchRsp(pRsp, rspSize, &batchRsp) < 0) {	37,013,536✔
157	code = TSDB_CODE_OUT_OF_MEMORY;	×
158	goto _exit;	×
159	}
160
161	_exit:	37,013,568✔
162
163	pMsg->info.rsp = pRsp;	37,013,757✔
164	pMsg->info.rspLen = rspSize;	37,013,523✔
165
166	if (code) {	37,013,791✔
167	mError("mnd get batch meta failed cause of %s", tstrerror(code));	×
168	}
169
170	taosArrayDestroyEx(batchReq.pMsgs, tFreeSBatchReqMsg);	37,013,791✔
171	taosArrayDestroyEx(batchRsp.pRsps, mnodeFreeSBatchRspMsg);	37,013,462✔
172
173	TAOS_RETURN(code);	37,013,649✔
174	}
175
176	int32_t mndInitQuery(SMnode *pMnode) {	527,264✔
177	if (qWorkerInit(NODE_TYPE_MNODE, MNODE_HANDLE, (void **)&pMnode->pQuery, &pMnode->msgCb) != 0) {	527,264✔
178	mError("failed to init qworker in mnode since %s", terrstr());	×
179	return -1;	×
180	}
181
182	mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY, mndProcessQueryMsg);	527,264✔
183	mndSetMsgHandleExt(pMnode, TDMT_SCH_MERGE_QUERY, mndProcessQueryMsg);	527,264✔
184	mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY_CONTINUE, mndProcessQueryMsg);	527,264✔
185	mndSetMsgHandleExt(pMnode, TDMT_SCH_FETCH, mndProcessQueryMsg);	527,264✔
186	mndSetMsgHandleExt(pMnode, TDMT_SCH_MERGE_FETCH, mndProcessQueryMsg);	527,264✔
187	mndSetMsgHandleExt(pMnode, TDMT_SCH_TASK_NOTIFY, mndProcessQueryMsg);	527,264✔
188	mndSetMsgHandleExt(pMnode, TDMT_SCH_DROP_TASK, mndProcessQueryMsg);	527,264✔
189	mndSetMsgHandleExt(pMnode, TDMT_SCH_QUERY_HEARTBEAT, mndProcessQueryMsg);	527,264✔
190	mndSetMsgHandle(pMnode, TDMT_MND_BATCH_META, mndProcessBatchMetaMsg);	527,264✔
191
192	return 0;	527,264✔
193	}
194
195	void mndCleanupQuery(SMnode pMnode) { qWorkerDestroy((void *)&pMnode->pQuery); }	527,202✔

taosdata / TDengine / #5056

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous