#5071

Committed 17 May 2026 01:15AM UTC coverage: 63.054% (-10.3%) from 73.326%

Build # #5071

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

238317 of 377957 relevant lines covered (63.05%)

130539817.12 hits per line

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

0.0

/source/dnode/vnode/src/vnd/vnodeScan.c

/*
 * Copyright (c) 2023 Hongze Cheng <hzcheng@umich.edu>.
 * All rights reserved.
 *
 * This code is the intellectual property of Hongze Cheng.
 * Any reproduction or distribution, in whole or in part,
 * without the express written permission of Hongze Cheng is
 * strictly prohibited.
 */

#include "vnd.h"
#include "vnode.h"
#include "vnodeInt.h"

extern int32_t tsdbAsyncScan(STsdb *tsdb, const STimeWindow *tw);

static int32_t vnodeAsyncScan(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
  SScanVnodeReq req = {0};

  int32_t code = tDeserializeSScanVnodeReq(pReq, len, &req);
  if (code) return code;

  vInfo("vgId:%d, scan msg will be processed, db:%s dbUid:%" PRId64 " scanStartTime:%" PRId64, TD_VID(pVnode), req.db,
        req.dbUid, req.scanStartTime);

  return tsdbAsyncScan(pVnode->pTsdb, &req.tw);
}

int32_t vnodeProcessScanVnodeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {
  if (!pVnode->restored) {
    vInfo("vgId:%d, ignore scan req during restoring. ver:%" PRId64, TD_VID(pVnode), ver);
    return 0;
  }
  return vnodeAsyncScan(pVnode, ver, pReq, len, pRsp);
}

extern void tsdbScanMonitorGetInfo(STsdb *tsdb, SQueryScanProgressRsp *rsp);

int32_t vnodeQueryScanProgress(SVnode *pVnode, SRpcMsg *pMsg) {
  int32_t code = 0;

  SQueryScanProgressReq req = {0};

  int32_t               rspSize = 0;
  SRpcMsg               rspMsg = {0};
  void                 *pRsp = NULL;
  SQueryScanProgressRsp rsp = {0};

  // deserialize request
  code = tDeserializeSQueryScanProgressReq(pMsg->pCont, pMsg->contLen, &req);
  if (code) {
    code = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

  // query scan progress
  rsp.dnodeId = req.dnodeId;
  tsdbScanMonitorGetInfo(pVnode->pTsdb, &rsp);
  vDebug("update scan progress, scanId:%d vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", rsp.scanId, rsp.vgId,
         rsp.dnodeId, rsp.numberFileset, rsp.finished);
  rsp.scanId = req.scanId;

  // serialize response
  rspSize = tSerializeSQueryScanProgressRsp(NULL, 0, &rsp);
  if (rspSize < 0) {
    code = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

  pRsp = rpcMallocCont(rspSize);
  if (pRsp == NULL) {
    vError("rpcMallocCont %d failed", rspSize);
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }
  code = tSerializeSQueryScanProgressRsp(pRsp, rspSize, &rsp);
  if (code < 0) {
    goto _exit;
  }
  code = 0;

_exit:
  rspMsg.info = pMsg->info;
  rspMsg.pCont = pRsp;
  rspMsg.contLen = rspSize;
  rspMsg.code = code;
  rspMsg.msgType = TDMT_VND_QUERY_COMPACT_PROGRESS_RSP;

  tmsgSendRsp(&rspMsg);

  return 0;
}

extern void tsdbCancelScanTask(STsdb *tsdb);

int32_t vnodeProcessKillScanReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {
  SVKillScanReq req = {0};

  vDebug("vgId:%d, kill scan msg will be processed, pReq:%p, len:%d", TD_VID(pVnode), pReq, len);
  int32_t code = tDeserializeSVKillScanReq(pReq, len, &req);
  if (code) {
    return TSDB_CODE_INVALID_MSG;
  }

  vInfo("vgId:%d, kill scan msg will be processed, scanId:%d, dnodeId:%d, vgId:%d", TD_VID(pVnode), req.scanId,
        req.dnodeId, req.vgId);

  tsdbCancelScanTask(pVnode->pTsdb);

  pRsp->msgType = TDMT_VND_KILL_SCAN_RSP;
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;

  return 0;
}

1	/*
2	* Copyright (c) 2023 Hongze Cheng <hzcheng@umich.edu>.
3	* All rights reserved.
4	*
5	* This code is the intellectual property of Hongze Cheng.
6	* Any reproduction or distribution, in whole or in part,
7	* without the express written permission of Hongze Cheng is
8	* strictly prohibited.
9	*/
10
11	#include "vnd.h"
12	#include "vnode.h"
13	#include "vnodeInt.h"
14
15	extern int32_t tsdbAsyncScan(STsdb tsdb, const STimeWindow tw);
16
17	static int32_t vnodeAsyncScan(SVnode pVnode, int64_t version, void pReq, int32_t len, SRpcMsg *pRsp) {	×
18	SScanVnodeReq req = {0};	×
19
20	int32_t code = tDeserializeSScanVnodeReq(pReq, len, &req);	×
21	if (code) return code;	×
22
23	vInfo("vgId:%d, scan msg will be processed, db:%s dbUid:%" PRId64 " scanStartTime:%" PRId64, TD_VID(pVnode), req.db,	×
24	req.dbUid, req.scanStartTime);
25
26	return tsdbAsyncScan(pVnode->pTsdb, &req.tw);	×
27	}
28
29	int32_t vnodeProcessScanVnodeReq(SVnode pVnode, int64_t ver, void pReq, int32_t len, SRpcMsg *pRsp) {	×
30	if (!pVnode->restored) {	×
31	vInfo("vgId:%d, ignore scan req during restoring. ver:%" PRId64, TD_VID(pVnode), ver);	×
32	return 0;	×
33	}
34	return vnodeAsyncScan(pVnode, ver, pReq, len, pRsp);	×
35	}
36
37	extern void tsdbScanMonitorGetInfo(STsdb tsdb, SQueryScanProgressRsp rsp);
38
39	int32_t vnodeQueryScanProgress(SVnode pVnode, SRpcMsg pMsg) {	×
40	int32_t code = 0;	×
41
42	SQueryScanProgressReq req = {0};	×
43
44	int32_t rspSize = 0;	×
45	SRpcMsg rspMsg = {0};	×
46	void *pRsp = NULL;	×
47	SQueryScanProgressRsp rsp = {0};	×
48
49	// deserialize request
50	code = tDeserializeSQueryScanProgressReq(pMsg->pCont, pMsg->contLen, &req);	×
51	if (code) {	×
52	code = TSDB_CODE_INVALID_MSG;	×
53	goto _exit;	×
54	}
55
56	// query scan progress
57	rsp.dnodeId = req.dnodeId;	×
58	tsdbScanMonitorGetInfo(pVnode->pTsdb, &rsp);	×
59	vDebug("update scan progress, scanId:%d vgId:%d, dnodeId:%d, numberFileset:%d, finished:%d", rsp.scanId, rsp.vgId,	×
60	rsp.dnodeId, rsp.numberFileset, rsp.finished);
61	rsp.scanId = req.scanId;	×
62
63	// serialize response
64	rspSize = tSerializeSQueryScanProgressRsp(NULL, 0, &rsp);	×
65	if (rspSize < 0) {	×
66	code = TSDB_CODE_INVALID_MSG;	×
67	goto _exit;	×
68	}
69
70	pRsp = rpcMallocCont(rspSize);	×
71	if (pRsp == NULL) {	×
72	vError("rpcMallocCont %d failed", rspSize);	×
73	code = TSDB_CODE_OUT_OF_MEMORY;	×
74	goto _exit;	×
75	}
76	code = tSerializeSQueryScanProgressRsp(pRsp, rspSize, &rsp);	×
77	if (code < 0) {	×
78	goto _exit;	×
79	}
80	code = 0;	×
81
82	_exit:	×
83	rspMsg.info = pMsg->info;	×
84	rspMsg.pCont = pRsp;	×
85	rspMsg.contLen = rspSize;	×
86	rspMsg.code = code;	×
87	rspMsg.msgType = TDMT_VND_QUERY_COMPACT_PROGRESS_RSP;	×
88
89	tmsgSendRsp(&rspMsg);	×
90
91	return 0;	×
92	}
93
94	extern void tsdbCancelScanTask(STsdb *tsdb);
95
96	int32_t vnodeProcessKillScanReq(SVnode pVnode, int64_t ver, void pReq, int32_t len, SRpcMsg *pRsp) {	×
97	SVKillScanReq req = {0};	×
98
99	vDebug("vgId:%d, kill scan msg will be processed, pReq:%p, len:%d", TD_VID(pVnode), pReq, len);	×
100	int32_t code = tDeserializeSVKillScanReq(pReq, len, &req);	×
101	if (code) {	×
102	return TSDB_CODE_INVALID_MSG;	×
103	}
104
105	vInfo("vgId:%d, kill scan msg will be processed, scanId:%d, dnodeId:%d, vgId:%d", TD_VID(pVnode), req.scanId,	×
106	req.dnodeId, req.vgId);
107
108	tsdbCancelScanTask(pVnode->pTsdb);	×
109
110	pRsp->msgType = TDMT_VND_KILL_SCAN_RSP;	×
111	pRsp->code = TSDB_CODE_SUCCESS;	×
112	pRsp->pCont = NULL;	×
113	pRsp->contLen = 0;	×
114
115	return 0;	×
116	}

taosdata / TDengine / #5071

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