#3525

Committed 10 Nov 2024 03:50AM UTC coverage: 60.818% (-0.08%) from 60.898%

Build # #3525

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28709 from taosdata/main

merge: from main to 3.0 branch

Run Details

118634 of 249004 branches covered (47.64%)

Branch coverage included in aggregate %.

136 of 169 new or added lines in 23 files covered. (80.47%)

542 existing lines in 129 files now uncovered.

199071 of 273386 relevant lines covered (72.82%)

15691647.46 hits per line

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

58.65

/source/dnode/vnode/src/tq/tqSnapshot.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 "meta.h"
#include "tdbInt.h"
#include "tq.h"

// STqSnapReader ========================================
struct STqSnapReader {
  STQ*    pTq;
  int64_t sver;
  int64_t ever;
  TBC*    pCur;
  int8_t  type;
};

int32_t tqSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, int8_t type, STqSnapReader** ppReader) {
  int32_t        code = 0;
  STqSnapReader* pReader = NULL;

  // alloc
  pReader = (STqSnapReader*)taosMemoryCalloc(1, sizeof(STqSnapReader));
  if (pReader == NULL) {
    code = terrno;
    goto _err;
  }
  pReader->pTq = pTq;
  pReader->sver = sver;
  pReader->ever = ever;
  pReader->type = type;

  // impl
  TTB* pTb = NULL;
  if (type == SNAP_DATA_TQ_CHECKINFO) {
    pTb = pTq->pCheckStore;
  } else if (type == SNAP_DATA_TQ_HANDLE) {
    pTb = pTq->pExecStore;
  } else if (type == SNAP_DATA_TQ_OFFSET) {
    pTb = pTq->pOffsetStore;
  } else {
    code = TSDB_CODE_INVALID_MSG;
    goto _err;
  }
  code = tdbTbcOpen(pTb, &pReader->pCur, NULL);
  if (code) {
    taosMemoryFree(pReader);
    goto _err;
  }

  code = tdbTbcMoveToFirst(pReader->pCur);
  if (code) {
    taosMemoryFree(pReader);
    goto _err;
  }

  tqInfo("vgId:%d, vnode snapshot tq reader opened", TD_VID(pTq->pVnode));

  *ppReader = pReader;
  return code;

_err:
  tqError("vgId:%d, vnode snapshot tq reader open failed since %s", TD_VID(pTq->pVnode), tstrerror(code));
  *ppReader = NULL;
  return code;
}

void tqSnapReaderClose(STqSnapReader** ppReader) {
  tdbTbcClose((*ppReader)->pCur);
  taosMemoryFree(*ppReader);
  *ppReader = NULL;
}

int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {
  int32_t code = 0;
  void*   pKey = NULL;
  void*   pVal = NULL;
  int32_t kLen = 0;
  int32_t vLen = 0;

  if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) {
    goto _exit;
  }

  *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);
  if (*ppData == NULL) {
    code = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
    goto _err;
  }

  SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
  pHdr->type = pReader->type;
  pHdr->size = vLen;
  (void)memcpy(pHdr->data, pVal, vLen);

_exit:
  tdbFree(pKey);
  tdbFree(pVal);
  tqInfo("vgId:%d, vnode snapshot tq read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);
  return code;

_err:
  tdbFree(pKey);
  tdbFree(pVal);
  tqError("vgId:%d, vnode snapshot tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code));
  return code;
}

// STqSnapWriter ========================================
struct STqSnapWriter {
  STQ*    pTq;
  int64_t sver;
  int64_t ever;
  TXN*    txn;
};

int32_t tqSnapWriterOpen(STQ* pTq, int64_t sver, int64_t ever, STqSnapWriter** ppWriter) {
  int32_t        code = 0;
  STqSnapWriter* pWriter = NULL;

  // alloc
  pWriter = (STqSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
  if (pWriter == NULL) {
    code = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);
    ;
    goto _err;
  }
  pWriter->pTq = pTq;
  pWriter->sver = sver;
  pWriter->ever = ever;

  code = tdbBegin(pTq->pMetaDB, &pWriter->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0);
  if (code < 0) {
    taosMemoryFree(pWriter);
    goto _err;
  }

  *ppWriter = pWriter;
  return code;

_err:
  tqError("vgId:%d, tq snapshot writer open failed since %s", TD_VID(pTq->pVnode), tstrerror(code));
  *ppWriter = NULL;
  return code;
}

int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
  int32_t        code = 0;
  STqSnapWriter* pWriter = *ppWriter;
  STQ*           pTq = pWriter->pTq;

  if (rollback) {
    tdbAbort(pWriter->pTq->pMetaDB, pWriter->txn);
  } else {
    code = tdbCommit(pWriter->pTq->pMetaDB, pWriter->txn);
    if (code) goto _err;
    code = tdbPostCommit(pWriter->pTq->pMetaDB, pWriter->txn);
    if (code) goto _err;
  }

  taosMemoryFree(pWriter);
  *ppWriter = NULL;

  return code;

_err:
  tqError("vgId:%d, tq snapshot writer close failed since %s", TD_VID(pTq->pVnode), tstrerror(code));
  return code;
}

int32_t tqSnapHandleWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
  int32_t   code = 0;
  STQ*      pTq = pWriter->pTq;
  SDecoder  decoder = {0};
  SDecoder* pDecoder = &decoder;
  STqHandle handle = {0};

  tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
  code = tDecodeSTqHandle(pDecoder, &handle);
  if (code) goto end;
  taosWLockLatch(&pTq->lock);
  code = tqMetaSaveInfo(pTq, pTq->pExecStore, handle.subKey, (int)strlen(handle.subKey), pData + sizeof(SSnapDataHdr),
                        nData - sizeof(SSnapDataHdr));
  taosWUnLockLatch(&pTq->lock);

end:
  tDecoderClear(pDecoder);
  tqDestroyTqHandle(&handle);
  tqInfo("vgId:%d, vnode snapshot tq write result:%d", TD_VID(pTq->pVnode), code);
  return code;
}

int32_t tqSnapCheckInfoWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
  int32_t      code = 0;
  STQ*         pTq = pWriter->pTq;
  STqCheckInfo info = {0};
  code = tqMetaDecodeCheckInfo(&info, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
  if (code != 0) {
    goto _err;
  }

  code = tqMetaSaveInfo(pTq, pTq->pCheckStore, &info.topic, strlen(info.topic), pData + sizeof(SSnapDataHdr),
                        nData - sizeof(SSnapDataHdr));
  tDeleteSTqCheckInfo(&info);
  if (code) goto _err;

  return code;

_err:
  tqError("vgId:%d, vnode check info tq write failed since %s", TD_VID(pTq->pVnode), tstrerror(code));
  return code;
}

int32_t tqSnapOffsetWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
  int32_t code = 0;
  STQ*    pTq = pWriter->pTq;

  STqOffset info = {0};
  code = tqMetaDecodeOffsetInfo(&info, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
  if (code != 0) {
    goto _err;
  }

  code = tqMetaSaveInfo(pTq, pTq->pOffsetStore, info.subKey, strlen(info.subKey), pData + sizeof(SSnapDataHdr),
                        nData - sizeof(SSnapDataHdr));
  tDeleteSTqOffset(&info);
  if (code) goto _err;

  return code;

_err:
  tqError("vgId:%d, vnode check info tq write failed since %s", TD_VID(pTq->pVnode), tstrerror(code));
  return code;
}

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 "meta.h"
17	#include "tdbInt.h"
18	#include "tq.h"
19
20	// STqSnapReader ========================================
21	struct STqSnapReader {
22	STQ* pTq;
23	int64_t sver;
24	int64_t ever;
25	TBC* pCur;
26	int8_t type;
27	};
28
29	int32_t tqSnapReaderOpen(STQ* pTq, int64_t sver, int64_t ever, int8_t type, STqSnapReader** ppReader) {	312✔
30	int32_t code = 0;	312✔
31	STqSnapReader* pReader = NULL;	312✔
32
33	// alloc
34	pReader = (STqSnapReader*)taosMemoryCalloc(1, sizeof(STqSnapReader));	312✔
35	if (pReader == NULL) {	312!
36	code = terrno;	×
37	goto _err;	×
38	}
39	pReader->pTq = pTq;	312✔
40	pReader->sver = sver;	312✔
41	pReader->ever = ever;	312✔
42	pReader->type = type;	312✔
43
44	// impl
45	TTB* pTb = NULL;	312✔
46	if (type == SNAP_DATA_TQ_CHECKINFO) {	312✔
47	pTb = pTq->pCheckStore;	104✔
48	} else if (type == SNAP_DATA_TQ_HANDLE) {	208✔
49	pTb = pTq->pExecStore;	104✔
50	} else if (type == SNAP_DATA_TQ_OFFSET) {	104!
51	pTb = pTq->pOffsetStore;	104✔
52	} else {
53	code = TSDB_CODE_INVALID_MSG;	×
54	goto _err;	×
55	}
56	code = tdbTbcOpen(pTb, &pReader->pCur, NULL);	312✔
57	if (code) {	312!
58	taosMemoryFree(pReader);	×
59	goto _err;	×
60	}
61
62	code = tdbTbcMoveToFirst(pReader->pCur);	312✔
63	if (code) {	312!
64	taosMemoryFree(pReader);	×
65	goto _err;	×
66	}
67
68	tqInfo("vgId:%d, vnode snapshot tq reader opened", TD_VID(pTq->pVnode));	312!
69
70	*ppReader = pReader;	312✔
71	return code;	312✔
72
73	_err:	×
74	tqError("vgId:%d, vnode snapshot tq reader open failed since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
75	*ppReader = NULL;	×
76	return code;	×
77	}
78
79	void tqSnapReaderClose(STqSnapReader** ppReader) {	312✔
80	tdbTbcClose((*ppReader)->pCur);	312✔
81	taosMemoryFree(*ppReader);	312✔
82	*ppReader = NULL;	312✔
83	}	312✔
84
85	int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {	397✔
86	int32_t code = 0;	397✔
87	void* pKey = NULL;	397✔
88	void* pVal = NULL;	397✔
89	int32_t kLen = 0;	397✔
90	int32_t vLen = 0;	397✔
91
92	if (tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen)) {	397✔
93	goto _exit;	312✔
94	}
95
96	*ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);	85✔
97	if (*ppData == NULL) {	85!
98	code = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);	×
99	goto _err;	×
100	}
101
102	SSnapDataHdr* pHdr = (SSnapDataHdr)(ppData);	85✔
103	pHdr->type = pReader->type;	85✔
104	pHdr->size = vLen;	85✔
105	(void)memcpy(pHdr->data, pVal, vLen);	85✔
106
107	_exit:	397✔
108	tdbFree(pKey);	397✔
109	tdbFree(pVal);	397✔
110	tqInfo("vgId:%d, vnode snapshot tq read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);	397!
111	return code;	397✔
112
113	_err:	×
114	tdbFree(pKey);	×
115	tdbFree(pVal);	×
116	tqError("vgId:%d, vnode snapshot tq read data failed since %s", TD_VID(pReader->pTq->pVnode), tstrerror(code));	×
117	return code;	×
118	}
119
120	// STqSnapWriter ========================================
121	struct STqSnapWriter {
122	STQ* pTq;
123	int64_t sver;
124	int64_t ever;
125	TXN* txn;
126	};
127
128	int32_t tqSnapWriterOpen(STQ* pTq, int64_t sver, int64_t ever, STqSnapWriter** ppWriter) {	85✔
129	int32_t code = 0;	85✔
130	STqSnapWriter* pWriter = NULL;	85✔
131
132	// alloc
133	pWriter = (STqSnapWriter)taosMemoryCalloc(1, sizeof(pWriter));	85✔
134	if (pWriter == NULL) {	85!
135	code = TAOS_GET_TERRNO(TSDB_CODE_OUT_OF_MEMORY);	×
136	;
137	goto _err;	×
138	}
139	pWriter->pTq = pTq;	85✔
140	pWriter->sver = sver;	85✔
141	pWriter->ever = ever;	85✔
142
143	code = tdbBegin(pTq->pMetaDB, &pWriter->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0);	85✔
144	if (code < 0) {	85!
145	taosMemoryFree(pWriter);	×
146	goto _err;	×
147	}
148
149	*ppWriter = pWriter;	85✔
150	return code;	85✔
151
152	_err:	×
153	tqError("vgId:%d, tq snapshot writer open failed since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
154	*ppWriter = NULL;	×
155	return code;	×
156	}
157
158	int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {	85✔
159	int32_t code = 0;	85✔
160	STqSnapWriter* pWriter = *ppWriter;	85✔
161	STQ* pTq = pWriter->pTq;	85✔
162
163	if (rollback) {	85!
164	tdbAbort(pWriter->pTq->pMetaDB, pWriter->txn);	×
165	} else {
166	code = tdbCommit(pWriter->pTq->pMetaDB, pWriter->txn);	85✔
167	if (code) goto _err;	85!
168	code = tdbPostCommit(pWriter->pTq->pMetaDB, pWriter->txn);	85✔
169	if (code) goto _err;	85!
170	}
171
172	taosMemoryFree(pWriter);	85✔
173	*ppWriter = NULL;	85✔
174
175	return code;	85✔
176
177	_err:	×
178	tqError("vgId:%d, tq snapshot writer close failed since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
179	return code;	×
180	}
181
182	int32_t tqSnapHandleWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {	43✔
183	int32_t code = 0;	43✔
184	STQ* pTq = pWriter->pTq;	43✔
185	SDecoder decoder = {0};	43✔
186	SDecoder* pDecoder = &decoder;	43✔
187	STqHandle handle = {0};	43✔
188
189	tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));	43✔
190	code = tDecodeSTqHandle(pDecoder, &handle);	43✔
191	if (code) goto end;	43!
192	taosWLockLatch(&pTq->lock);	43✔
193	code = tqMetaSaveInfo(pTq, pTq->pExecStore, handle.subKey, (int)strlen(handle.subKey), pData + sizeof(SSnapDataHdr),	43✔
194	nData - sizeof(SSnapDataHdr));	43✔
195	taosWUnLockLatch(&pTq->lock);	43✔
196
197	end:	43✔
198	tDecoderClear(pDecoder);	43✔
199	tqDestroyTqHandle(&handle);	43✔
200	tqInfo("vgId:%d, vnode snapshot tq write result:%d", TD_VID(pTq->pVnode), code);	43!
201	return code;	43✔
202	}
203
UNCOV 204	int32_t tqSnapCheckInfoWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {	×
UNCOV 205	int32_t code = 0;	×
UNCOV 206	STQ* pTq = pWriter->pTq;	×
UNCOV 207	STqCheckInfo info = {0};	×
UNCOV 208	code = tqMetaDecodeCheckInfo(&info, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));	×
UNCOV 209	if (code != 0) {	×
210	goto _err;	×
211	}
212
UNCOV 213	code = tqMetaSaveInfo(pTq, pTq->pCheckStore, &info.topic, strlen(info.topic), pData + sizeof(SSnapDataHdr),	×
UNCOV 214	nData - sizeof(SSnapDataHdr));	×
UNCOV 215	tDeleteSTqCheckInfo(&info);	×
UNCOV 216	if (code) goto _err;	×
217
UNCOV 218	return code;	×
219
220	_err:	×
221	tqError("vgId:%d, vnode check info tq write failed since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
222	return code;	×
223	}
224
225	int32_t tqSnapOffsetWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {	42✔
226	int32_t code = 0;	42✔
227	STQ* pTq = pWriter->pTq;	42✔
228
229	STqOffset info = {0};	42✔
230	code = tqMetaDecodeOffsetInfo(&info, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));	42✔
231	if (code != 0) {	42!
232	goto _err;	×
233	}
234
235	code = tqMetaSaveInfo(pTq, pTq->pOffsetStore, info.subKey, strlen(info.subKey), pData + sizeof(SSnapDataHdr),	42✔
236	nData - sizeof(SSnapDataHdr));	42✔
237	tDeleteSTqOffset(&info);	42✔
238	if (code) goto _err;	42!
239
240	return code;	42✔
241
242	_err:	×
243	tqError("vgId:%d, vnode check info tq write failed since %s", TD_VID(pTq->pVnode), tstrerror(code));	×
244	return code;	×
245	}

taosdata / TDengine / #3525

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