#3653

Committed 14 Mar 2025 08:10AM UTC coverage: 22.565% (-41.0%) from 63.596%

Build # #3653

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(keep): support keep on super table level. (#30097)

* Feat: support use keep while create super table.

* Test(keep): add test for create super table with keep option.

* Feat(keep): Add tmsg for create keep.

* Feat(keep): support alter table option keep.

* Fix(keep): Add baisc test for alter table option.

* Fix(keep): memory leek.

* Feat(keep): add keep to metaEntry&metaCache and fix earliestTs with stn keep.

* Test(keep): add some cases for select with stb keep.

* Fix: fix ci core while alter stb.

* Feat(keep): delete expired data in super table level.

* Feat: remove get stb keep while query.

* Fix : build error.

* Revert "Fix : build error."

This reverts commit 0ed66e4e8.

* Revert "Feat(keep): delete expired data in super table level."

This reverts commit 36330f6b4.

* Fix : build errors.

* Feat : support restart taosd.

* Fix : alter table comment problems.

* Test : add tests for super table keep.

* Fix: change sdb stb reserve size.

* Test: add more tests.

* Feat: Disable normal tables and sub tables from setting the keep parameter

* Fix: add more checks to avoid unknown address.

* Docs: Add docs for stable keep.

* Fix: some review changes.

* Fix: review errors.

Run Details

49248 of 302527 branches covered (16.28%)

Branch coverage included in aggregate %.

53 of 99 new or added lines in 12 files covered. (53.54%)

155872 existing lines in 443 files now uncovered.

87359 of 302857 relevant lines covered (28.84%)

570004.22 hits per line

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

0.0

/source/dnode/vnode/src/tq/tqScan.c

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 "tq.h"
UNCOV 17	static int32_t tqAddRawDataToRsp(const void* rawData, SMqDataRsp* pRsp, int8_t precision) {	×
UNCOV 18	int32_t code = TDB_CODE_SUCCESS;	×
UNCOV 19	int32_t lino = 0;	×
UNCOV 20	void* buf = NULL;	×
21
UNCOV 22	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + (uint32_t )rawData + INT_BYTES;	×
UNCOV 23	buf = taosMemoryCalloc(1, dataStrLen);	×
UNCOV 24	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	×
25
UNCOV 26	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	×
UNCOV 27	pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_RAW_VERSION;	×
UNCOV 28	pRetrieve->precision = precision;	×
UNCOV 29	pRetrieve->compressed = 0;	×
30
UNCOV 31	memcpy(pRetrieve->data, rawData, (uint32_t )rawData + INT_BYTES);	×
UNCOV 32	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &dataStrLen), code, lino, END, terrno);	×
UNCOV 33	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	×
UNCOV 34	pRsp->blockDataElementFree = true;	×
35
UNCOV 36	tqTrace("tqAddRawDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);	×
UNCOV 37	END:	×
UNCOV 38	if (code != TSDB_CODE_SUCCESS) {	×
39	taosMemoryFree(buf);	×
40	tqError("%s failed at %d, failed to add block data to response:%s", __FUNCTION__, lino, tstrerror(code));	×
41	}
UNCOV 42	return code;	×
43	}
44
UNCOV 45	static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision) {	×
UNCOV 46	int32_t code = 0;	×
UNCOV 47	int32_t lino = 0;	×
48
UNCOV 49	size_t dataEncodeBufSize = blockGetEncodeSize(pBlock);	×
UNCOV 50	int32_t dataStrLen = sizeof(SRetrieveTableRspForTmq) + dataEncodeBufSize;	×
UNCOV 51	void* buf = taosMemoryCalloc(1, dataStrLen);	×
UNCOV 52	TSDB_CHECK_NULL(buf, code, lino, END, terrno);	×
53
UNCOV 54	SRetrieveTableRspForTmq* pRetrieve = (SRetrieveTableRspForTmq*)buf;	×
UNCOV 55	pRetrieve->version = RETRIEVE_TABLE_RSP_TMQ_VERSION;	×
UNCOV 56	pRetrieve->precision = precision;	×
UNCOV 57	pRetrieve->compressed = 0;	×
UNCOV 58	pRetrieve->numOfRows = htobe64((int64_t)pBlock->info.rows);	×
59
UNCOV 60	int32_t actualLen = blockEncode(pBlock, pRetrieve->data, dataEncodeBufSize, numOfCols);	×
UNCOV 61	TSDB_CHECK_CONDITION(actualLen >= 0, code, lino, END, terrno);	×
62
UNCOV 63	actualLen += sizeof(SRetrieveTableRspForTmq);	×
UNCOV 64	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockDataLen, &actualLen), code, lino, END, terrno);	×
UNCOV 65	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockData, &buf), code, lino, END, terrno);	×
UNCOV 66	pRsp->blockDataElementFree = true;	×
UNCOV 67	tqTrace("tqAddBlockDataToRsp add block data to block array, blockDataLen:%d, blockData:%p", dataStrLen, buf);	×
68
UNCOV 69	END:	×
UNCOV 70	if (code != TSDB_CODE_SUCCESS){	×
71	taosMemoryFree(buf);	×
72	tqError("%s failed at line %d with msg:%s", __func__, lino, tstrerror(code));	×
73	}
UNCOV 74	return code;	×
75	}
76
UNCOV 77	static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, SMqDataRsp* pRsp, int32_t n) {	×
UNCOV 78	int32_t code = TDB_CODE_SUCCESS;	×
UNCOV 79	int32_t lino = 0;	×
UNCOV 80	SMetaReader mr = {0};	×
81
UNCOV 82	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
UNCOV 83	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
84
UNCOV 85	metaReaderDoInit(&mr, pTq->pVnode->pMeta, META_READER_LOCK);	×
86
UNCOV 87	code = metaReaderGetTableEntryByUidCache(&mr, uid);	×
UNCOV 88	TSDB_CHECK_CODE(code, lino, END);	×
89
UNCOV 90	for (int32_t i = 0; i < n; i++) {	×
UNCOV 91	char* tbName = taosStrdup(mr.me.name);	×
UNCOV 92	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	×
UNCOV 93	if(taosArrayPush(pRsp->blockTbName, &tbName) == NULL){	×
94	tqError("failed to push tbName to blockTbName:%s, uid:%"PRId64, tbName, uid);	×
95	continue;	×
96	}
UNCOV 97	tqTrace("add tbName to response success tbname:%s, uid:%"PRId64, tbName, uid);	×
98	}
99
UNCOV 100	END:	×
UNCOV 101	if (code != TSDB_CODE_SUCCESS) {	×
UNCOV 102	tqError("%s failed at %d, failed to add tbName to response:%s, uid:%"PRId64, __FUNCTION__, lino, tstrerror(code), uid);	×
103	}
UNCOV 104	metaReaderClear(&mr);	×
UNCOV 105	return code;	×
106	}
107
UNCOV 108	int32_t getDataBlock(qTaskInfo_t task, const STqHandle* pHandle, int32_t vgId, SSDataBlock** res) {	×
UNCOV 109	if (task == NULL \|\| pHandle == NULL \|\| res == NULL) {	×
110	return TSDB_CODE_INVALID_PARA;	×
111	}
UNCOV 112	uint64_t ts = 0;	×
UNCOV 113	qStreamSetOpen(task);	×
114
UNCOV 115	tqDebug("consumer:0x%" PRIx64 " vgId:%d, tmq one task start execute", pHandle->consumerId, vgId);	×
UNCOV 116	int32_t code = qExecTask(task, res, &ts);	×
UNCOV 117	if (code != TSDB_CODE_SUCCESS) {	×
118	tqError("consumer:0x%" PRIx64 " vgId:%d, task exec error since %s", pHandle->consumerId, vgId, tstrerror(code));	×
119	return code;	×
120	}
121
UNCOV 122	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq one task end executed, pDataBlock:%p", pHandle->consumerId, vgId, *res);	×
UNCOV 123	return 0;	×
124	}
125
UNCOV 126	static int32_t tqProcessReplayRsp(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, const SMqPollReq* pRequest, SSDataBlock* pDataBlock, qTaskInfo_t task){	×
UNCOV 127	int32_t code = 0;	×
UNCOV 128	int32_t lino = 0;	×
129
UNCOV 130	if (IS_OFFSET_RESET_TYPE(pRequest->reqOffset.type) && pHandle->block != NULL) {	×
UNCOV 131	blockDataDestroy(pHandle->block);	×
UNCOV 132	pHandle->block = NULL;	×
133	}
UNCOV 134	if (pHandle->block == NULL) {	×
UNCOV 135	if (pDataBlock == NULL) {	×
UNCOV 136	goto END;	×
137	}
138
UNCOV 139	STqOffsetVal offset = {0};	×
UNCOV 140	code = qStreamExtractOffset(task, &offset);	×
UNCOV 141	TSDB_CHECK_CODE(code, lino, END);	×
142
UNCOV 143	pHandle->block = NULL;	×
144
UNCOV 145	code = createOneDataBlock(pDataBlock, true, &pHandle->block);	×
UNCOV 146	TSDB_CHECK_CODE(code, lino, END);	×
147
UNCOV 148	pHandle->blockTime = offset.ts;	×
UNCOV 149	tOffsetDestroy(&offset);	×
UNCOV 150	int32_t vgId = TD_VID(pTq->pVnode);	×
UNCOV 151	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	×
UNCOV 152	TSDB_CHECK_CODE(code, lino, END);	×
153	}
154
UNCOV 155	const STqExecHandle* pExec = &pHandle->execHandle;	×
UNCOV 156	code = tqAddBlockDataToRsp(pHandle->block, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);	×
UNCOV 157	TSDB_CHECK_CODE(code, lino, END);	×
158
UNCOV 159	pRsp->blockNum++;	×
UNCOV 160	if (pDataBlock == NULL) {	×
UNCOV 161	blockDataDestroy(pHandle->block);	×
UNCOV 162	pHandle->block = NULL;	×
163	} else {
UNCOV 164	code = copyDataBlock(pHandle->block, pDataBlock);	×
UNCOV 165	TSDB_CHECK_CODE(code, lino, END);	×
166
UNCOV 167	STqOffsetVal offset = {0};	×
UNCOV 168	code = qStreamExtractOffset(task, &offset);	×
UNCOV 169	TSDB_CHECK_CODE(code, lino, END);	×
170
UNCOV 171	pRsp->sleepTime = offset.ts - pHandle->blockTime;	×
UNCOV 172	pHandle->blockTime = offset.ts;	×
UNCOV 173	tOffsetDestroy(&offset);	×
174	}
175
UNCOV 176	END:	×
UNCOV 177	if (code != TSDB_CODE_SUCCESS) {	×
178	tqError("%s failed at %d, failed to process replay response:%s", __FUNCTION__, lino, tstrerror(code));	×
179	}
UNCOV 180	return code;	×
181	}
182
UNCOV 183	int32_t tqScanData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	×
UNCOV 184	int32_t code = 0;	×
UNCOV 185	int32_t lino = 0;	×
UNCOV 186	TSDB_CHECK_NULL(pRsp, code, lino, END, TSDB_CODE_INVALID_PARA);	×
UNCOV 187	TSDB_CHECK_NULL(pTq, code, lino, END, TSDB_CODE_INVALID_PARA);	×
UNCOV 188	TSDB_CHECK_NULL(pHandle, code, lino, END, TSDB_CODE_INVALID_PARA);	×
UNCOV 189	TSDB_CHECK_NULL(pOffset, code, lino, END, TSDB_CODE_INVALID_PARA);	×
UNCOV 190	TSDB_CHECK_NULL(pRequest, code, lino, END, TSDB_CODE_INVALID_PARA);	×
191
UNCOV 192	int32_t vgId = TD_VID(pTq->pVnode);	×
UNCOV 193	int32_t totalRows = 0;	×
194
UNCOV 195	const STqExecHandle* pExec = &pHandle->execHandle;	×
UNCOV 196	qTaskInfo_t task = pExec->task;	×
197
UNCOV 198	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	×
UNCOV 199	TSDB_CHECK_CODE(code, lino, END);	×
200
UNCOV 201	qStreamSetSourceExcluded(task, pRequest->sourceExcluded);	×
UNCOV 202	int64_t st = taosGetTimestampMs();	×
UNCOV 203	while (1) {	×
UNCOV 204	SSDataBlock* pDataBlock = NULL;	×
UNCOV 205	code = getDataBlock(task, pHandle, vgId, &pDataBlock);	×
UNCOV 206	TSDB_CHECK_CODE(code, lino, END);	×
207
UNCOV 208	if (pRequest->enableReplay) {	×
UNCOV 209	code = tqProcessReplayRsp(pTq, pHandle, pRsp, pRequest, pDataBlock, task);	×
UNCOV 210	TSDB_CHECK_CODE(code, lino, END);	×
UNCOV 211	break;	×
212	}
UNCOV 213	if (pDataBlock == NULL) {	×
UNCOV 214	break;	×
215	}
UNCOV 216	code = tqAddBlockDataToRsp(pDataBlock, pRsp, pExec->numOfCols, pTq->pVnode->config.tsdbCfg.precision);	×
UNCOV 217	TSDB_CHECK_CODE(code, lino, END);	×
218
UNCOV 219	pRsp->blockNum++;	×
UNCOV 220	totalRows += pDataBlock->info.rows;	×
UNCOV 221	if (totalRows >= pRequest->minPollRows \|\| (taosGetTimestampMs() - st > pRequest->timeout)) {	×
222	break;
223	}
224	}
225
UNCOV 226	tqDebug("consumer:0x%" PRIx64 " vgId:%d tmq task executed finished, total blocks:%d, totalRows:%d", pHandle->consumerId, vgId, pRsp->blockNum, totalRows);	×
UNCOV 227	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
228
UNCOV 229	END:	×
UNCOV 230	if (code != 0) {	×
UNCOV 231	tqError("%s failed at %d, tmq task executed error msg:%s", __FUNCTION__, lino, tstrerror(code));	×
232	}
UNCOV 233	return code;	×
234	}
235
UNCOV 236	int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, SMqBatchMetaRsp* pBatchMetaRsp, STqOffsetVal* pOffset, const SMqPollReq* pRequest) {	×
UNCOV 237	int32_t code = 0;	×
UNCOV 238	int32_t lino = 0;	×
UNCOV 239	char* tbName = NULL;	×
UNCOV 240	SSchemaWrapper* pSW = NULL;	×
UNCOV 241	const STqExecHandle* pExec = &pHandle->execHandle;	×
UNCOV 242	qTaskInfo_t task = pExec->task;	×
UNCOV 243	code = qStreamPrepareScan(task, pOffset, pHandle->execHandle.subType);	×
UNCOV 244	TSDB_CHECK_CODE(code, lino, END);	×
245
UNCOV 246	int32_t rowCnt = 0;	×
UNCOV 247	int64_t st = taosGetTimestampMs();	×
UNCOV 248	while (1) {	×
UNCOV 249	SSDataBlock* pDataBlock = NULL;	×
UNCOV 250	uint64_t ts = 0;	×
UNCOV 251	tqDebug("tmqsnap task start to execute");	×
UNCOV 252	code = qExecTask(task, &pDataBlock, &ts);	×
UNCOV 253	TSDB_CHECK_CODE(code, lino, END);	×
UNCOV 254	tqDebug("tmqsnap task execute end, get %p", pDataBlock);	×
255
UNCOV 256	if (pDataBlock != NULL && pDataBlock->info.rows > 0) {	×
UNCOV 257	if (pRsp->withTbName) {	×
UNCOV 258	tbName = taosStrdup(qExtractTbnameFromTask(task));	×
UNCOV 259	TSDB_CHECK_NULL(tbName, code, lino, END, terrno);	×
UNCOV 260	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockTbName, &tbName), code, lino, END, terrno);	×
UNCOV 261	tqDebug("vgId:%d, add tbname:%s to rsp msg", pTq->pVnode->config.vgId, tbName);	×
UNCOV 262	tbName = NULL;	×
263	}
UNCOV 264	if (pRsp->withSchema) {	×
UNCOV 265	pSW = tCloneSSchemaWrapper(qExtractSchemaFromTask(task));	×
UNCOV 266	TSDB_CHECK_NULL(pSW, code, lino, END, terrno);	×
UNCOV 267	TSDB_CHECK_NULL(taosArrayPush(pRsp->blockSchema, &pSW), code, lino, END, terrno);	×
UNCOV 268	pSW = NULL;	×
269	}
270
UNCOV 271	code = tqAddBlockDataToRsp(pDataBlock, pRsp, taosArrayGetSize(pDataBlock->pDataBlock),	×
UNCOV 272	pTq->pVnode->config.tsdbCfg.precision);	×
UNCOV 273	TSDB_CHECK_CODE(code, lino, END);	×
274
UNCOV 275	pRsp->blockNum++;	×
UNCOV 276	rowCnt += pDataBlock->info.rows;	×
UNCOV 277	if (rowCnt <= pRequest->minPollRows && (taosGetTimestampMs() - st <= pRequest->timeout)) {	×
UNCOV 278	continue;	×
279	}
280	}
281
282	// get meta
UNCOV 283	SMqBatchMetaRsp* tmp = qStreamExtractMetaMsg(task);	×
UNCOV 284	if (taosArrayGetSize(tmp->batchMetaReq) > 0) {	×
UNCOV 285	code = qStreamExtractOffset(task, &tmp->rspOffset);	×
UNCOV 286	TSDB_CHECK_CODE(code, lino, END);	×
UNCOV 287	pBatchMetaRsp = tmp;	×
UNCOV 288	tqDebug("tmqsnap task get meta");	×
UNCOV 289	break;	×
290	}
291
UNCOV 292	if (pDataBlock == NULL) {	×
UNCOV 293	code = qStreamExtractOffset(task, pOffset);	×
UNCOV 294	TSDB_CHECK_CODE(code, lino, END);	×
295
UNCOV 296	if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {	×
UNCOV 297	continue;	×
298	}
299
UNCOV 300	tqDebug("tmqsnap vgId: %d, tsdb consume over, switch to wal, ver %" PRId64, TD_VID(pTq->pVnode), pHandle->snapshotVer + 1);	×
UNCOV 301	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
UNCOV 302	break;	×
303	}
304
UNCOV 305	if (pRsp->blockNum > 0) {	×
UNCOV 306	tqDebug("tmqsnap task exec exited, get data");	×
UNCOV 307	code = qStreamExtractOffset(task, &pRsp->rspOffset);	×
UNCOV 308	break;	×
309	}
310	}
UNCOV 311	tqDebug("%s:%d success", __FUNCTION__, lino);	×
312	END:	×
UNCOV 313	if (code != 0){	×
UNCOV 314	tqError("%s failed at %d, vgId:%d, task exec error since %s", __FUNCTION__ , lino, pTq->pVnode->config.vgId, tstrerror(code));	×
315	}
UNCOV 316	taosMemoryFree(pSW);	×
UNCOV 317	taosMemoryFree(tbName);	×
UNCOV 318	return code;	×
319	}
320
UNCOV 321	static void tqProcessSubData(STQ* pTq, STqHandle* pHandle, SMqDataRsp* pRsp, int32_t* totalRows,	×
322	const SMqPollReq* pRequest, SArray* rawList){
UNCOV 323	int32_t code = 0;	×
UNCOV 324	int32_t lino = 0;	×
UNCOV 325	SArray* pBlocks = NULL;	×
UNCOV 326	SArray* pSchemas = NULL;	×
327
UNCOV 328	STqExecHandle* pExec = &pHandle->execHandle;	×
UNCOV 329	STqReader* pReader = pExec->pTqReader;	×
330
UNCOV 331	pBlocks = taosArrayInit(0, sizeof(SSDataBlock));	×
UNCOV 332	TSDB_CHECK_NULL(pBlocks, code, lino, END, terrno);	×
UNCOV 333	pSchemas = taosArrayInit(0, sizeof(void*));	×
UNCOV 334	TSDB_CHECK_NULL(pSchemas, code, lino, END, terrno);	×
335
UNCOV 336	SSubmitTbData* pSubmitTbData = NULL;	×
UNCOV 337	code = tqRetrieveTaosxBlock(pReader, pRsp, pBlocks, pSchemas, &pSubmitTbData, rawList, pHandle->fetchMeta);	×
UNCOV 338	TSDB_CHECK_CODE(code, lino, END);	×
UNCOV 339	bool tmp = (pSubmitTbData->flags & pRequest->sourceExcluded) != 0;	×
UNCOV 340	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	×
341
UNCOV 342	if (pHandle->fetchMeta == ONLY_META){	×
UNCOV 343	goto END;	×
344	}
345
UNCOV 346	int32_t blockNum = taosArrayGetSize(pBlocks) == 0 ? 1 : taosArrayGetSize(pBlocks);	×
UNCOV 347	if (pRsp->withTbName) {	×
UNCOV 348	int64_t uid = pExec->pTqReader->lastBlkUid;	×
UNCOV 349	code = tqAddTbNameToRsp(pTq, uid, pRsp, blockNum);	×
UNCOV 350	TSDB_CHECK_CODE(code, lino, END);	×
351	}
352
UNCOV 353	TSDB_CHECK_CONDITION(!tmp, code, lino, END, TSDB_CODE_SUCCESS);	×
UNCOV 354	for (int32_t i = 0; i < blockNum; i++) {	×
UNCOV 355	if (taosArrayGetSize(pBlocks) == 0){	×
UNCOV 356	void* rawData = taosArrayGetP(rawList, pReader->nextBlk - 1);	×
UNCOV 357	if (rawData == NULL) {	×
358	continue;	×
359	}
UNCOV 360	if (tqAddRawDataToRsp(rawData, pRsp, pTq->pVnode->config.tsdbCfg.precision) != 0){	×
361	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
362	continue;	×
363	}
UNCOV 364	totalRows += (uint32_t *)rawData + INT_BYTES; // bytes actually	×
365	} else {
UNCOV 366	SSDataBlock* pBlock = taosArrayGet(pBlocks, i);	×
UNCOV 367	if (pBlock == NULL) {	×
368	continue;	×
369	}
370
UNCOV 371	if (tqAddBlockDataToRsp(pBlock, pRsp, taosArrayGetSize(pBlock->pDataBlock), pTq->pVnode->config.tsdbCfg.precision) != 0){	×
372	tqError("vgId:%d, failed to add block to rsp msg", pTq->pVnode->config.vgId);	×
373	continue;	×
374	}
UNCOV 375	*totalRows += pBlock->info.rows;	×
376	}
377
UNCOV 378	void** pSW = taosArrayGet(pSchemas, i);	×
UNCOV 379	if (taosArrayPush(pRsp->blockSchema, pSW) == NULL){	×
380	tqError("vgId:%d, failed to add schema to rsp msg", pTq->pVnode->config.vgId);	×
381	continue;	×
382	}
UNCOV 383	*pSW = NULL;	×
UNCOV 384	pRsp->blockNum++;	×
385	}
UNCOV 386	tqTrace("vgId:%d, process sub data success, response blocknum:%d, rows:%d", pTq->pVnode->config.vgId, pRsp->blockNum, *totalRows);	×
UNCOV 387	END:	×
UNCOV 388	if (code != 0) {	×
UNCOV 389	tqError("%s failed at %d, failed to process sub data:%s", __FUNCTION__, lino, tstrerror(code));	×
390	}
UNCOV 391	taosArrayDestroyEx(pBlocks, (FDelete)blockDataFreeRes);	×
UNCOV 392	taosArrayDestroyP(pSchemas, (FDelete)tDeleteSchemaWrapper);	×
UNCOV 393	}	×
394
UNCOV 395	static void preProcessSubmitMsg(STqHandle* pHandle, const SMqPollReq* pRequest, SArray** rawList){	×
UNCOV 396	STqExecHandle* pExec = &pHandle->execHandle;	×
UNCOV 397	STqReader* pReader = pExec->pTqReader;	×
UNCOV 398	int32_t blockSz = taosArrayGetSize(pReader->submit.aSubmitTbData);	×
UNCOV 399	for (int32_t i = 0; i < blockSz; i++){	×
UNCOV 400	SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, i);	×
UNCOV 401	if (pSubmitTbData== NULL){	×
402	taosArrayDestroy(*rawList);	×
403	*rawList = NULL;	×
UNCOV 404	return;	×
405	}
406
UNCOV 407	int64_t uid = pSubmitTbData->uid;	×
UNCOV 408	if (pRequest->rawData) {	×
UNCOV 409	if (taosHashGet(pRequest->uidHash, &uid, LONG_BYTES) != NULL) {	×
UNCOV 410	tqDebug("poll rawdata split,uid:%" PRId64 " is already exists", uid);	×
UNCOV 411	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
UNCOV 412	return;	×
413	} else {
UNCOV 414	int32_t code = taosHashPut(pRequest->uidHash, &uid, LONG_BYTES, &uid, LONG_BYTES);	×
UNCOV 415	if (code != 0) {	×
416	tqError("failed to add table uid to hash, code:%d, uid:%" PRId64, code, uid);	×
417	}
418	}
419	}
420
UNCOV 421	if (pSubmitTbData->pCreateTbReq == NULL){	×
UNCOV 422	continue;	×
423	}
424
UNCOV 425	int64_t createTime = INT64_MAX;	×
UNCOV 426	int64_t cTime = (int64_t)taosHashGet(pHandle->tableCreateTimeHash, &uid, LONG_BYTES);	×
UNCOV 427	if (cTime != NULL){	×
UNCOV 428	createTime = *cTime;	×
429	} else{
UNCOV 430	createTime = metaGetTableCreateTime(pReader->pVnodeMeta, uid, 1);	×
UNCOV 431	if (createTime != INT64_MAX){	×
UNCOV 432	int32_t code = taosHashPut(pHandle->tableCreateTimeHash, &uid, LONG_BYTES, &createTime, LONG_BYTES);	×
UNCOV 433	if (code != 0){	×
434	tqError("failed to add table create time to hash,code:%d, uid:%"PRId64, code, uid);	×
435	}
436	}
437	}
UNCOV 438	if (pHandle->fetchMeta == WITH_DATA \|\| pSubmitTbData->ctimeMs > createTime){	×
UNCOV 439	tDestroySVCreateTbReq(pSubmitTbData->pCreateTbReq, TSDB_MSG_FLG_DECODE);	×
UNCOV 440	pSubmitTbData->pCreateTbReq = NULL;	×
441	} else{
UNCOV 442	taosArrayDestroy(*rawList);	×
UNCOV 443	*rawList = NULL;	×
444	}
445	}
446	}
447
UNCOV 448	int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, SMqDataRsp* pRsp, int32_t* totalRows, const SMqPollReq* pRequest) {	×
UNCOV 449	int32_t code = 0;	×
UNCOV 450	int32_t lino = 0;	×
UNCOV 451	STqExecHandle* pExec = &pHandle->execHandle;	×
UNCOV 452	STqReader* pReader = pExec->pTqReader;	×
UNCOV 453	SArray *rawList = NULL;	×
UNCOV 454	if (pRequest->rawData){	×
UNCOV 455	rawList = taosArrayInit(0, POINTER_BYTES);	×
UNCOV 456	TSDB_CHECK_NULL(rawList, code, lino, END, terrno);	×
457	}
UNCOV 458	code = tqReaderSetSubmitMsg(pReader, submit.msgStr, submit.msgLen, submit.ver, rawList);	×
UNCOV 459	TSDB_CHECK_CODE(code, lino, END);	×
UNCOV 460	preProcessSubmitMsg(pHandle, pRequest, &rawList);	×
461	// data could not contains same uid data in rawdata mode
UNCOV 462	if (pRequest->rawData != 0 && terrno == TSDB_CODE_TMQ_RAW_DATA_SPLIT){	×
UNCOV 463	goto END;	×
464	}
465
466	// this submit data is metadata and previous data is rawdata
UNCOV 467	if (pRequest->rawData != 0 && *totalRows > 0 && pRsp->createTableNum == 0 && rawList == NULL){	×
468	tqDebug("poll rawdata split,vgId:%d, this wal submit data contains metadata and previous data is data", pTq->pVnode->config.vgId);	×
469	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
470	goto END;	×
471	}
472
473	// this submit data is rawdata and previous data is metadata
UNCOV 474	if (pRequest->rawData != 0 && pRsp->createTableNum > 0 && rawList != NULL){	×
475	tqDebug("poll rawdata split,vgId:%d, this wal submit data is data and previous data is metadata", pTq->pVnode->config.vgId);	×
476	terrno = TSDB_CODE_TMQ_RAW_DATA_SPLIT;	×
477	goto END;	×
478	}
479
UNCOV 480	if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {	×
UNCOV 481	while (tqNextBlockImpl(pReader, NULL)) {	×
UNCOV 482	tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);	×
483	}
UNCOV 484	} else if (pExec->subType == TOPIC_SUB_TYPE__DB) {	×
UNCOV 485	while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {	×
UNCOV 486	tqProcessSubData(pTq, pHandle, pRsp, totalRows, pRequest, rawList);	×
487	}
488	}
489
UNCOV 490	END:	×
UNCOV 491	tqReaderClearSubmitMsg(pReader);	×
UNCOV 492	taosArrayDestroy(rawList);	×
UNCOV 493	if (code != 0){	×
494	tqError("%s failed at %d, failed to scan log:%s", __FUNCTION__, lino, tstrerror(code));	×
495	}
UNCOV 496	return code;	×
497	}

taosdata / TDengine / #3653

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