#5134

Committed 03 Nov 2026 01:50AM UTC coverage: 58.984% (-11.3%) from 70.274%

Build # #5134

Build Type

push

travis-ci

Committed by jbjia

Commit Message

fix: run coverage workers on 1.46 instead of the CI host

Coverage and the tsdb CI were sharing 192.168.1.47, so database leases and taosd contended for the same machine. Point the coverage machine list at 192.168.1.46.

Co-authored-by: Cursor <cursoragent@cursor.com>

Coverage Stats

294921 of 499999 relevant lines covered (58.98%)

7457186.83 hits per line

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

76.0

/source/dnode/qnode/src/qnode.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * SPDX-License-Identifier: AGPL-3.0-or-later
 *
 * 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 "tqueue.h"

#include "executor.h"
#include "qndInt.h"
#include "query.h"
#include "qworker.h"

int32_t qndOpen(const SQnodeOpt *pOption, SQnode **pQnode) {
  *pQnode = taosMemoryCalloc(1, sizeof(SQnode));
  if (NULL == *pQnode) {
    qError("calloc SQnode failed");
    return terrno;
  }
  (*pQnode)->qndId = QNODE_HANDLE;

  int32_t code = qWorkerInit(NODE_TYPE_QNODE, (*pQnode)->qndId, (void **)&(*pQnode)->pQuery, &pOption->msgCb);
  if (TSDB_CODE_SUCCESS != code) {
    taosMemoryFreeClear(pQnode);
    return code;
  }

  (*pQnode)->msgCb = pOption->msgCb;
  return TSDB_CODE_SUCCESS;
}

void qndClose(SQnode *pQnode) {
  qWorkerDestroy((void **)&pQnode->pQuery);
  taosMemoryFree(pQnode);
}

int32_t qndGetLoad(SQnode *pQnode, SQnodeLoad *pLoad) {
  SReadHandle  handle = {.pMsgCb = &pQnode->msgCb};
  SQWorkerStat stat = {0};

  int32_t code = qWorkerGetStat(&handle, pQnode->pQuery, &stat);
  if (code) {
    return code;
  }

  pLoad->numOfQueryInQueue = stat.numOfQueryInQueue;
  pLoad->numOfFetchInQueue = stat.numOfFetchInQueue;
  pLoad->timeInQueryQueue = stat.timeInQueryQueue;
  pLoad->timeInFetchQueue = stat.timeInFetchQueue;
  pLoad->cacheDataSize = stat.cacheDataSize;
  pLoad->numOfProcessedQuery = stat.queryProcessed;
  pLoad->numOfProcessedCQuery = stat.cqueryProcessed;
  pLoad->numOfProcessedFetch = stat.fetchProcessed;
  pLoad->numOfProcessedDrop = stat.dropProcessed;
  pLoad->numOfProcessedNotify = stat.notifyProcessed;
  pLoad->numOfProcessedHb = stat.hbProcessed;
  pLoad->numOfProcessedDelete = stat.deleteProcessed;

  return 0;
}

int32_t qndPreprocessQueryMsg(SQnode *pQnode, SRpcMsg *pMsg) {
  int32_t qType = 0;
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {
    return 0;
  }

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

int32_t qndProcessQueryMsg(SQnode *pQnode, SQueueInfo *pInfo, SRpcMsg *pMsg) {
  int32_t     code = -1;
  int64_t     ts = pInfo->timestamp;
  SReadHandle handle = {0};
  handle.pMsgCb = &pQnode->msgCb;
  handle.pWorkerCb = pInfo->workerCb;
  
  const STraceId *trace = &pMsg->info.traceId;
  qDebug("qnodeId:%d, msg:%s, %p in qnode queue is processing, %" PRIx64 ":%" PRIx64, 
    pQnode->qndId, TMSG_INFO(pMsg->msgType), pMsg, TRACE_GET_ROOTID(trace), TRACE_GET_MSGID(trace));

  switch (pMsg->msgType) {
    case TDMT_SCH_QUERY:
    case TDMT_SCH_MERGE_QUERY:
      code = qWorkerProcessQueryMsg(&handle, pQnode->pQuery, pMsg, ts);
      break;
    case TDMT_SCH_QUERY_CONTINUE:
      code = qWorkerProcessCQueryMsg(&handle, pQnode->pQuery, pMsg, ts);
      break;
    case TDMT_SCH_FETCH:
    case TDMT_SCH_MERGE_FETCH:
      code = qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg, ts);
      break;
    case TDMT_SCH_CANCEL_TASK:
      // code = qWorkerProcessCancelMsg(pQnode, pQnode->pQuery, pMsg, ts);
      break;
    case TDMT_SCH_DROP_TASK:
      code = qWorkerProcessDropMsg(pQnode, pQnode->pQuery, pMsg, ts);
      break;
    case TDMT_VND_TMQ_CONSUME:
      // code =  tqProcessConsumeReq(pQnode->pTq, pMsg);
      // break;
    case TDMT_SCH_QUERY_HEARTBEAT:
      code = qWorkerProcessHbMsg(pQnode, pQnode->pQuery, pMsg, ts);
      break;
    case TDMT_SCH_TASK_NOTIFY:
      code = qWorkerProcessNotifyMsg(pQnode, pQnode->pQuery, pMsg, ts);
      break;
    default:
      qError("unknown msg type:%d in qnode queue", pMsg->msgType);
      terrno = TSDB_CODE_APP_ERROR;
  }

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

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* SPDX-License-Identifier: AGPL-3.0-or-later
5	*
6	* This program is free software: you can use, redistribute, and/or modify
7	* it under the terms of the GNU Affero General Public License, version 3
8	* or later ("AGPL"), as published by the Free Software Foundation.
9	*
10	* This program is distributed in the hope that it will be useful, but WITHOUT
11	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12	* FITNESS FOR A PARTICULAR PURPOSE.
13	*
14	* You should have received a copy of the GNU Affero General Public License
15	* along with this program. If not, see <http://www.gnu.org/licenses/>.
16	*/
17
18	#include "tqueue.h"
19
20	#include "executor.h"
21	#include "qndInt.h"
22	#include "query.h"
23	#include "qworker.h"
24
25	int32_t qndOpen(const SQnodeOpt pOption, SQnode *pQnode) {	1,147 ✔
26	*pQnode = taosMemoryCalloc(1, sizeof(SQnode));	1,147 ✔
27	if (NULL == *pQnode) {	1,147 ✔
28	qError("calloc SQnode failed");	×
29	return terrno;	×
30	}
31	(*pQnode)->qndId = QNODE_HANDLE;	1,147 ✔
32
33	int32_t code = qWorkerInit(NODE_TYPE_QNODE, (pQnode)->qndId, (void )&(pQnode)->pQuery, &pOption->msgCb);	1,147 ✔
34	if (TSDB_CODE_SUCCESS != code) {	1,147 ✔
35	taosMemoryFreeClear(pQnode);	×
36	return code;	×
37	}
38
39	(*pQnode)->msgCb = pOption->msgCb;	1,147 ✔
40	return TSDB_CODE_SUCCESS;	1,147 ✔
41	}
42
43	void qndClose(SQnode *pQnode) {	1,147 ✔
44	qWorkerDestroy((void **)&pQnode->pQuery);	1,147 ✔
45	taosMemoryFree(pQnode);	1,147 ✔
46	}	1,147 ✔
47
48	int32_t qndGetLoad(SQnode pQnode, SQnodeLoad pLoad) {	85,987 ✔
49	SReadHandle handle = {.pMsgCb = &pQnode->msgCb};	85,987 ✔
50	SQWorkerStat stat = {0};	85,987 ✔
51
52	int32_t code = qWorkerGetStat(&handle, pQnode->pQuery, &stat);	85,987 ✔
53	if (code) {	85,987 ✔
54	return code;	×
55	}
56
57	pLoad->numOfQueryInQueue = stat.numOfQueryInQueue;	85,987 ✔
58	pLoad->numOfFetchInQueue = stat.numOfFetchInQueue;	85,987 ✔
59	pLoad->timeInQueryQueue = stat.timeInQueryQueue;	85,987 ✔
60	pLoad->timeInFetchQueue = stat.timeInFetchQueue;	85,987 ✔
61	pLoad->cacheDataSize = stat.cacheDataSize;	85,987 ✔
62	pLoad->numOfProcessedQuery = stat.queryProcessed;	85,987 ✔
63	pLoad->numOfProcessedCQuery = stat.cqueryProcessed;	85,987 ✔
64	pLoad->numOfProcessedFetch = stat.fetchProcessed;	85,987 ✔
65	pLoad->numOfProcessedDrop = stat.dropProcessed;	85,987 ✔
66	pLoad->numOfProcessedNotify = stat.notifyProcessed;	85,987 ✔
67	pLoad->numOfProcessedHb = stat.hbProcessed;	85,987 ✔
68	pLoad->numOfProcessedDelete = stat.deleteProcessed;	85,987 ✔
69
70	return 0;	85,987 ✔
71	}
72
73	int32_t qndPreprocessQueryMsg(SQnode pQnode, SRpcMsg pMsg) {	3,140 ✔
74	int32_t qType = 0;	3,140 ✔
75	if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {	3,140 ✔
76	return 0;	×
77	}
78
79	return qWorkerPreprocessQueryMsg(pQnode->pQuery, pMsg, false, &qType);	3,140 ✔
80	}
81
82	int32_t qndProcessQueryMsg(SQnode pQnode, SQueueInfo pInfo, SRpcMsg *pMsg) {	11,806 ✔
83	int32_t code = -1;	11,806 ✔
84	int64_t ts = pInfo->timestamp;	11,806 ✔
85	SReadHandle handle = {0};	11,806 ✔
86	handle.pMsgCb = &pQnode->msgCb;	11,806 ✔
87	handle.pWorkerCb = pInfo->workerCb;	11,806 ✔
88
89	const STraceId *trace = &pMsg->info.traceId;	11,806 ✔
90	qDebug("qnodeId:%d, msg:%s, %p in qnode queue is processing, %" PRIx64 ":%" PRIx64,	11,806 ✔
91	pQnode->qndId, TMSG_INFO(pMsg->msgType), pMsg, TRACE_GET_ROOTID(trace), TRACE_GET_MSGID(trace));
92
93	switch (pMsg->msgType) {	11,806 ✔
94	case TDMT_SCH_QUERY:	3,140 ✔
95	case TDMT_SCH_MERGE_QUERY:
96	code = qWorkerProcessQueryMsg(&handle, pQnode->pQuery, pMsg, ts);	3,140 ✔
97	break;	3,140 ✔
98	case TDMT_SCH_QUERY_CONTINUE:	×
99	code = qWorkerProcessCQueryMsg(&handle, pQnode->pQuery, pMsg, ts);	×
100	break;	×
101	case TDMT_SCH_FETCH:	2,733 ✔
102	case TDMT_SCH_MERGE_FETCH:
103	code = qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg, ts);	2,733 ✔
104	break;	2,733 ✔
105	case TDMT_SCH_CANCEL_TASK:	×
106	// code = qWorkerProcessCancelMsg(pQnode, pQnode->pQuery, pMsg, ts);
107	break;	×
108	case TDMT_SCH_DROP_TASK:	3,140 ✔
109	code = qWorkerProcessDropMsg(pQnode, pQnode->pQuery, pMsg, ts);	3,140 ✔
110	break;	3,140 ✔
111	case TDMT_VND_TMQ_CONSUME:	2,793 ✔
112	// code = tqProcessConsumeReq(pQnode->pTq, pMsg);
113	// break;
114	case TDMT_SCH_QUERY_HEARTBEAT:
115	code = qWorkerProcessHbMsg(pQnode, pQnode->pQuery, pMsg, ts);	2,793 ✔
116	break;	2,793 ✔
117	case TDMT_SCH_TASK_NOTIFY:	×
118	code = qWorkerProcessNotifyMsg(pQnode, pQnode->pQuery, pMsg, ts);	×
119	break;	×
120	default:	×
121	qError("unknown msg type:%d in qnode queue", pMsg->msgType);	×
122	terrno = TSDB_CODE_APP_ERROR;	×
123	}
124
125	if (code == 0) {	11,806 ✔
126	return TSDB_CODE_ACTION_IN_PROGRESS;	11,746 ✔
127	} else {
128	return code;	×
129	}
130	}

taosdata / TDengine / #5134

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