• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #3638

11 Mar 2025 12:59PM UTC coverage: 3.066% (-18.3%) from 21.409%
#3638

push

travis-ci

web-flow
Merge pull request #30118 from taosdata/wl30

udpate ci workflow

5914 of 287117 branches covered (2.06%)

Branch coverage included in aggregate %.

11588 of 283747 relevant lines covered (4.08%)

142.17 hits per line

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

0.0
/source/dnode/vnode/src/tq/tqSnapshot.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 "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) {
×
30
  int      code  = TSDB_CODE_SUCCESS;
×
31
  int32_t  lino  = 0;
×
32
  STqSnapReader* pReader = NULL;
×
33
  TSDB_CHECK_NULL(pTq, code, lino, end, TSDB_CODE_INVALID_MSG);
×
34
  TSDB_CHECK_NULL(ppReader, code, lino, end, TSDB_CODE_INVALID_MSG);
×
35

36
  // alloc
37
  pReader = (STqSnapReader*)taosMemoryCalloc(1, sizeof(STqSnapReader));
×
38
  TSDB_CHECK_NULL(pReader, code, lino, end, terrno);
×
39

40
  pReader->pTq = pTq;
×
41
  pReader->sver = sver;
×
42
  pReader->ever = ever;
×
43
  pReader->type = type;
×
44

45
  // impl
46
  TTB* pTb = NULL;
×
47
  if (type == SNAP_DATA_TQ_CHECKINFO) {
×
48
    pTb = pTq->pCheckStore;
×
49
  } else if (type == SNAP_DATA_TQ_HANDLE) {
×
50
    pTb = pTq->pExecStore;
×
51
  } else if (type == SNAP_DATA_TQ_OFFSET) {
×
52
    pTb = pTq->pOffsetStore;
×
53
  } else {
54
    code = TSDB_CODE_INVALID_MSG;
×
55
    goto end;
×
56
  }
57
  code = tdbTbcOpen(pTb, &pReader->pCur, NULL);
×
58
  TSDB_CHECK_CODE(code, lino, end);
×
59
  code = tdbTbcMoveToFirst(pReader->pCur);
×
60
  TSDB_CHECK_CODE(code, lino, end);
×
61
  tqInfo("vgId:%d, vnode tq snapshot reader opene success", TD_VID(pTq->pVnode));
×
62
  *ppReader = pReader;
×
63

64
end:
×
65
  if (code != 0){
×
66
    tqError("%s failed at %d, vnode tq snapshot reader open failed since %s", __func__, lino, tstrerror(code));
×
67
    taosMemoryFreeClear(pReader);
×
68
  }
69

70
  return code;
×
71
}
72

73
void tqSnapReaderClose(STqSnapReader** ppReader) {
×
74
  if (ppReader == NULL || *ppReader == NULL) {
×
75
    return;
×
76
  }
77
  tdbTbcClose((*ppReader)->pCur);
×
78
  taosMemoryFreeClear(*ppReader);
×
79
}
80

81
int32_t tqSnapRead(STqSnapReader* pReader, uint8_t** ppData) {
×
82
  int     code  = TSDB_CODE_SUCCESS;
×
83
  int32_t lino  = 0;
×
84
  void*   pKey  = NULL;
×
85
  void*   pVal  = NULL;
×
86
  int32_t kLen  = 0;
×
87
  int32_t vLen  = 0;
×
88
  TSDB_CHECK_NULL(pReader, code, lino, end, TSDB_CODE_INVALID_MSG);
×
89
  TSDB_CHECK_NULL(ppData, code, lino, end, TSDB_CODE_INVALID_MSG);
×
90

91
  code = tdbTbcNext(pReader->pCur, &pKey, &kLen, &pVal, &vLen);
×
92
  TSDB_CHECK_CONDITION(code == 0, code, lino, end, TDB_CODE_SUCCESS);
×
93

94
  *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + vLen);
×
95
  TSDB_CHECK_NULL(*ppData, code, lino, end, terrno);
×
96

97
  SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
×
98
  pHdr->type = pReader->type;
×
99
  pHdr->size = vLen;
×
100
  (void)memcpy(pHdr->data, pVal, vLen);
×
101
  tqInfo("vgId:%d, vnode tq snapshot read data, vLen:%d", TD_VID(pReader->pTq->pVnode), vLen);
×
102

103
end:
×
104
  if (code != 0) {
×
105
    tqError("%s failed at %d, vnode tq snapshot read data failed since %s", __func__, lino, tstrerror(code));
×
106
  }
107
  tdbFree(pKey);
×
108
  tdbFree(pVal);
×
109
  return code;
×
110
}
111

112
// STqSnapWriter ========================================
113
struct STqSnapWriter {
114
  STQ*    pTq;
115
  int64_t sver;
116
  int64_t ever;
117
  TXN*    txn;
118
};
119

120
int32_t tqSnapWriterOpen(STQ* pTq, int64_t sver, int64_t ever, STqSnapWriter** ppWriter) {
×
121
  int     code  = TSDB_CODE_SUCCESS;
×
122
  int32_t lino  = 0;
×
123
  STqSnapWriter* pWriter = NULL;
×
124

125
  TSDB_CHECK_NULL(pTq, code, lino, end, TSDB_CODE_INVALID_MSG);
×
126
  TSDB_CHECK_NULL(ppWriter, code, lino, end, TSDB_CODE_INVALID_MSG);
×
127

128
  // alloc
129
  pWriter = (STqSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
×
130
  TSDB_CHECK_NULL(pWriter, code, lino, end, terrno);
×
131
  pWriter->pTq = pTq;
×
132
  pWriter->sver = sver;
×
133
  pWriter->ever = ever;
×
134

135
  code = tdbBegin(pTq->pMetaDB, &pWriter->txn, tdbDefaultMalloc, tdbDefaultFree, NULL, 0);
×
136
  TSDB_CHECK_CODE(code, lino, end);
×
137
  tqInfo("vgId:%d, tq snapshot writer opene success", TD_VID(pTq->pVnode));
×
138
  *ppWriter = pWriter;
×
139

140
end:
×
141
  if (code != 0){
×
142
    tqError("%s failed at %d tq snapshot writer open failed since %s", __func__, lino, tstrerror(code));
×
143
    taosMemoryFreeClear(pWriter);
×
144
  }
145
  return code;
×
146
}
147

148
int32_t tqSnapWriterClose(STqSnapWriter** ppWriter, int8_t rollback) {
×
149
  int     code  = TSDB_CODE_SUCCESS;
×
150
  int32_t lino  = 0;
×
151
  STqSnapWriter* pWriter = NULL;
×
152

153
  TSDB_CHECK_NULL(ppWriter, code, lino, end, TSDB_CODE_INVALID_MSG);
×
154
  TSDB_CHECK_NULL(*ppWriter, code, lino, end, TSDB_CODE_INVALID_MSG);
×
155
  pWriter = *ppWriter;
×
156

157
  if (rollback) {
×
158
    tdbAbort(pWriter->pTq->pMetaDB, pWriter->txn);
×
159
  } else {
160
    code = tdbCommit(pWriter->pTq->pMetaDB, pWriter->txn);
×
161
    TSDB_CHECK_CODE(code, lino, end);
×
162

163
    code = tdbPostCommit(pWriter->pTq->pMetaDB, pWriter->txn);
×
164
    TSDB_CHECK_CODE(code, lino, end);
×
165
  }
166
  tqInfo("vgId:%d, tq snapshot writer close success", TD_VID(pWriter->pTq->pVnode));
×
167
  taosMemoryFreeClear(*ppWriter);
×
168

169
end:
×
170
  if (code != 0){
×
171
    tqError("%s failed at %d, tq snapshot writer close failed since %s", __func__, lino, tstrerror(code));
×
172
  }
173
  return code;
×
174
}
175

176
static int32_t tqWriteCheck(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData){
×
177
  int       code  = TSDB_CODE_SUCCESS;
×
178
  int32_t   lino  = 0;
×
179
  TSDB_CHECK_NULL(pWriter, code, lino, end, TSDB_CODE_INVALID_MSG);
×
180
  TSDB_CHECK_NULL(pData, code, lino, end, TSDB_CODE_INVALID_MSG);
×
181
  TSDB_CHECK_CONDITION(nData >= sizeof(SSnapDataHdr), code, lino, end, TSDB_CODE_INVALID_MSG);
×
182
end:
×
183
  if (code != 0){
×
184
    tqError("%s failed at %d failed since %s", __func__, lino, tstrerror(code));
×
185
  }
186
  return code;
×
187
}
188
int32_t tqSnapHandleWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
×
189
  int       code  = TSDB_CODE_SUCCESS;
×
190
  int32_t   lino  = 0;
×
191
  SDecoder  decoder = {0};
×
192
  SDecoder* pDecoder = &decoder;
×
193
  STqHandle handle = {0};
×
194
  code = tqWriteCheck(pWriter, pData, nData);
×
195
  TSDB_CHECK_CODE(code, lino, end);
×
196

197
  STQ*      pTq = pWriter->pTq;
×
198
  tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
×
199
  code = tDecodeSTqHandle(pDecoder, &handle);
×
200
  TSDB_CHECK_CODE(code, lino, end);
×
201

202
  taosWLockLatch(&pTq->lock);
×
203
  code = tqMetaSaveInfo(pTq, pTq->pExecStore, handle.subKey, strlen(handle.subKey), pData + sizeof(SSnapDataHdr),
×
204
                        nData - sizeof(SSnapDataHdr));
205
  taosWUnLockLatch(&pTq->lock);
×
206
  TSDB_CHECK_CODE(code, lino, end);
×
207
  tqInfo("vgId:%d, vnode tq snapshot write success", TD_VID(pTq->pVnode));
×
208

209
end:
×
210
  tDecoderClear(pDecoder);
×
211
  tqDestroyTqHandle(&handle);
×
212
  if (code != 0){
×
213
    tqError("%s failed at %d, vnode tq snapshot write failed since %s", __func__, lino, tstrerror(code));
×
214
  }
215
  return code;
×
216
}
217

218
int32_t tqSnapCheckInfoWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
×
219
  int       code  = TSDB_CODE_SUCCESS;
×
220
  int32_t   lino  = 0;
×
221

222
  code = tqWriteCheck(pWriter, pData, nData);
×
223
  TSDB_CHECK_CODE(code, lino, end);
×
224

225
  STQ*         pTq = pWriter->pTq;
×
226
  STqCheckInfo info = {0};
×
227
  code = tqMetaDecodeCheckInfo(&info, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
×
228
  TSDB_CHECK_CODE(code, lino, end);
×
229

230
  code = tqMetaSaveInfo(pTq, pTq->pCheckStore, &info.topic, strlen(info.topic), pData + sizeof(SSnapDataHdr),
×
231
                        nData - sizeof(SSnapDataHdr));
232
  tDeleteSTqCheckInfo(&info);
×
233
  TSDB_CHECK_CODE(code, lino, end);
×
234
  tqInfo("vgId:%d, vnode tq check info  write success", TD_VID(pTq->pVnode));
×
235

236
end:
×
237
  if (code != 0){
×
238
    tqError("%s failed at %d, vnode tq check info write failed since %s", __func__, lino, tstrerror(code));
×
239
  }
240
  return code;
×
241
}
242

243
int32_t tqSnapOffsetWrite(STqSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
×
244
  int       code  = TSDB_CODE_SUCCESS;
×
245
  int32_t   lino  = 0;
×
246
  code = tqWriteCheck(pWriter, pData, nData);
×
247
  TSDB_CHECK_CODE(code, lino, end);
×
248

249
  STQ*    pTq = pWriter->pTq;
×
250
  STqOffset info = {0};
×
251
  code = tqMetaDecodeOffsetInfo(&info, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
×
252
  TSDB_CHECK_CODE(code, lino, end);
×
253

254
  code = tqMetaSaveInfo(pTq, pTq->pOffsetStore, info.subKey, strlen(info.subKey), pData + sizeof(SSnapDataHdr),
×
255
                        nData - sizeof(SSnapDataHdr));
256
  tDeleteSTqOffset(&info);
×
257
  TSDB_CHECK_CODE(code, lino, end);
×
258
  tqInfo("vgId:%d, vnode tq offset write success", TD_VID(pTq->pVnode));
×
259

260
end:
×
261
  if (code != 0){
×
262
    tqError("%s failed at %d, vnode tq offset write failed since %s", __func__, lino, tstrerror(code));
×
263
  }
264
  return code;
×
265
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc