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

taosdata / TDengine / #3778

28 Mar 2025 06:51AM UTC coverage: 63.374% (+0.4%) from 62.934%
#3778

push

travis-ci

web-flow
fix(tdb): disable page recycling (#30529)

155771 of 313582 branches covered (49.67%)

Branch coverage included in aggregate %.

241569 of 313390 relevant lines covered (77.08%)

20705705.27 hits per line

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

74.01
/source/dnode/vnode/src/tsdb/tsdbCacheRead.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 "functionMgt.h"
17
#include "functionResInfo.h"
18
#include "taoserror.h"
19
#include "tarray.h"
20
#include "tcommon.h"
21
#include "tsdb.h"
22
#include "tsdbDataFileRW.h"
23
#include "tsdbReadUtil.h"
24

25
#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
26

27
static int32_t setFirstLastResColToNull(SColumnInfoData* pCol, int32_t row) {
44✔
28
  int32_t        code = TSDB_CODE_SUCCESS;
44✔
29
  int32_t        lino = 0;
44✔
30
  char*          buf = NULL;
44✔
31
  SFirstLastRes* pRes = NULL;
44✔
32

33
  TSDB_CHECK_NULL(pCol, code, lino, _end, TSDB_CODE_INVALID_PARA);
44!
34

35
  buf = taosMemoryCalloc(1, pCol->info.bytes);
44!
36
  TSDB_CHECK_NULL(buf, code, lino, _end, terrno);
44!
37

38
  pRes = (SFirstLastRes*)((char*)buf + VARSTR_HEADER_SIZE);
44✔
39
  pRes->bytes = 0;
44✔
40
  pRes->hasResult = true;
44✔
41
  pRes->isNull = true;
44✔
42
  varDataSetLen(buf, pCol->info.bytes - VARSTR_HEADER_SIZE);
44✔
43
  code = colDataSetVal(pCol, row, buf, false);
44✔
44
  TSDB_CHECK_CODE(code, lino, _end);
44!
45

46
_end:
44✔
47
  if (code != TSDB_CODE_SUCCESS) {
44!
48
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
49
  }
50
  if (buf != NULL) {
44!
51
    taosMemoryFreeClear(buf);
44!
52
  }
53
  return code;
44✔
54
}
55

56
static int32_t saveOneRowForLastRaw(SLastCol* pColVal, SCacheRowsReader* pReader, const int32_t slotId,
10,522✔
57
                                    SColumnInfoData* pColInfoData, int32_t numOfRows) {
58
  int32_t  code = TSDB_CODE_SUCCESS;
10,522✔
59
  int32_t  lino = 0;
10,522✔
60
  SColVal* pVal = NULL;
10,522✔
61

62
  TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,522!
63
  TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,522!
64

65
  pVal = &pColVal->colVal;
10,522✔
66

67
  // allNullRow = false;
68
  if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {
10,522!
69
    if (!COL_VAL_IS_VALUE(&pColVal->colVal)) {
994✔
70
      colDataSetNULL(pColInfoData, numOfRows);
128✔
71
    } else {
72
      TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
866!
73
      varDataSetLen(pReader->transferBuf[slotId], pVal->value.nData);
866✔
74

75
      memcpy(varDataVal(pReader->transferBuf[slotId]), pVal->value.pData, pVal->value.nData);
866✔
76
      code = colDataSetVal(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);
866✔
77
      TSDB_CHECK_CODE(code, lino, _end);
871!
78
    }
79
  } else {
80
    code = colDataSetVal(pColInfoData, numOfRows, VALUE_GET_DATUM(&pVal->value, pColVal->colVal.value.type),
9,528!
81
                         !COL_VAL_IS_VALUE(pVal));
9,528!
82
    TSDB_CHECK_CODE(code, lino, _end);
9,530!
83
  }
84

85
_end:
9,530✔
86
  if (code != TSDB_CODE_SUCCESS) {
10,529!
87
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
88
  }
89
  return code;
10,528✔
90
}
91

92
static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,
10,178✔
93
                          const int32_t* dstSlotIds, void** pRes, const char* idStr) {
94
  int32_t code = TSDB_CODE_SUCCESS;
10,178✔
95
  int32_t lino = 0;
10,178✔
96
  int32_t numOfRows = 0;
10,178✔
97
  SArray* funcTypeBlockArray = NULL;
10,178✔
98

99
  TSDB_CHECK_NULL(pBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,178!
100
  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,178!
101
  if (pReader->numOfCols > 0) {
10,178✔
102
    TSDB_CHECK_NULL(slotIds, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,177!
103
    TSDB_CHECK_NULL(dstSlotIds, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,177!
104
    TSDB_CHECK_NULL(pRes, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,177!
105
  }
106

107
  numOfRows = pBlock->info.rows;
10,178✔
108

109
  if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {
10,178✔
110
    uint64_t       ts = TSKEY_MIN;
5,229✔
111
    SFirstLastRes* p = NULL;
5,229✔
112
    col_id_t       colId = -1;
5,229✔
113

114
    funcTypeBlockArray = taosArrayInit(pReader->numOfCols, sizeof(int32_t));
5,229✔
115
    TSDB_CHECK_NULL(funcTypeBlockArray, code, lino, _end, terrno);
5,227!
116

117
    for (int32_t i = 0; i < pReader->numOfCols; ++i) {
13,029✔
118
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
7,799✔
119
      TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);
7,799!
120

121
      int32_t funcType = FUNCTION_TYPE_CACHE_LAST;
7,799✔
122
      if (pReader->pFuncTypeList != NULL && taosArrayGetSize(pReader->pFuncTypeList) > i) {
7,799✔
123
        void* pVal = taosArrayGet(pReader->pFuncTypeList, i);
1,572✔
124
        TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
1,572!
125

126
        funcType = *(int32_t*)pVal;
1,572✔
127
        pVal = taosArrayGet(pReader->pFuncTypeList, i);
1,572✔
128
        TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
1,572!
129

130
        void* px = taosArrayInsert(funcTypeBlockArray, dstSlotIds[i], pVal);
1,572✔
131
        TSDB_CHECK_NULL(px, code, lino, _end, terrno);
1,572✔
132
      }
133

134
      if (slotIds[i] == -1) {
7,798✔
135
        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
44!
136
          colDataSetNULL(pColInfoData, numOfRows);
×
137
          continue;
×
138
        }
139

140
        code = setFirstLastResColToNull(pColInfoData, numOfRows);
44✔
141
        TSDB_CHECK_CODE(code, lino, _end);
44!
142
        continue;
44✔
143
      }
144

145
      int32_t   slotId = slotIds[i];
7,754✔
146
      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
7,754✔
147
      TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
7,753!
148

149
      colId = pColVal->colVal.cid;
7,753✔
150
      if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
7,753✔
151
        code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);
826✔
152
        TSDB_CHECK_CODE(code, lino, _end);
826!
153

154
        continue;
826✔
155
      }
156

157
      p = (SFirstLastRes*)varDataVal(pRes[i]);
6,927✔
158

159
      p->ts = pColVal->rowKey.ts;
6,927✔
160
      ts = p->ts;
6,927✔
161
      p->isNull = !COL_VAL_IS_VALUE(&pColVal->colVal);
6,927✔
162
      // allNullRow = p->isNull & allNullRow;
163
      if (!p->isNull) {
6,927✔
164
        if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {
6,915!
165
          varDataSetLen(p->buf, pColVal->colVal.value.nData);
1,437✔
166

167
          memcpy(varDataVal(p->buf), pColVal->colVal.value.pData, pColVal->colVal.value.nData);
1,437✔
168
          p->bytes = pColVal->colVal.value.nData + VARSTR_HEADER_SIZE;  // binary needs to plus the header size
1,437✔
169
        } else {
170
          memcpy(p->buf, VALUE_GET_DATUM(&pColVal->colVal.value, pColVal->colVal.value.type),
5,478!
171
                 pReader->pSchema->columns[slotId].bytes);
5,478✔
172
          p->bytes = pReader->pSchema->columns[slotId].bytes;
5,478✔
173
        }
174
      }
175

176
      // pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to subtract it
177
      p->hasResult = true;
6,927✔
178
      varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);
6,927✔
179
      code = colDataSetVal(pColInfoData, numOfRows, (const char*)pRes[i], false);
6,927✔
180
      TSDB_CHECK_CODE(code, lino, _end);
6,932!
181
    }
182

183
    for (int32_t idx = 0; idx < taosArrayGetSize(pBlock->pDataBlock); ++idx) {
20,315✔
184
      SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, idx);
15,085✔
185
      TSDB_CHECK_NULL(pCol, code, lino, _end, TSDB_CODE_INVALID_PARA);
15,085!
186

187
      if (idx < funcTypeBlockArray->size) {
15,085✔
188
        void* pVal = taosArrayGet(funcTypeBlockArray, idx);
1,573✔
189
        TSDB_CHECK_NULL(pVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
1,573!
190

191
        int32_t funcType = *(int32_t*)pVal;
1,573✔
192
        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
1,573✔
193
          continue;
826✔
194
        }
195
      }
196

197
      if (pCol->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID && pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
14,259✔
198
        if (ts == TSKEY_MIN) {
5,874✔
199
          colDataSetNULL(pCol, numOfRows);
44!
200
        } else {
201
          code = colDataSetVal(pCol, numOfRows, (const char*)&ts, false);
5,830✔
202
          TSDB_CHECK_CODE(code, lino, _end);
5,830!
203
        }
204
        continue;
5,874✔
205
      } else if (pReader->numOfCols == 1 && idx != dstSlotIds[0] && (pCol->info.colId == colId || colId == -1)) {
8,385✔
206
        if (p && !p->isNull) {
1,196!
207
          code = colDataSetVal(pCol, numOfRows, p->buf, false);
1,152✔
208
          TSDB_CHECK_CODE(code, lino, _end);
1,152!
209
        } else {
210
          colDataSetNULL(pCol, numOfRows);
44!
211
        }
212
      }
213
    }
214

215
    // pBlock->info.rows += allNullRow ? 0 : 1;
216
    ++pBlock->info.rows;
5,224✔
217
  } else if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST_ROW)) {
4,949!
218
    for (int32_t i = 0; i < pReader->numOfCols; ++i) {
15,000✔
219
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
10,048✔
220
      TSDB_CHECK_NULL(pColInfoData, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,046!
221

222
      int32_t slotId = slotIds[i];
10,046✔
223
      if (slotId == -1) {
10,046✔
224
        colDataSetNULL(pColInfoData, numOfRows);
349!
225
        continue;
349✔
226
      }
227

228
      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
9,697✔
229
      TSDB_CHECK_NULL(pColVal, code, lino, _end, TSDB_CODE_INVALID_PARA);
9,698!
230

231
      code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);
9,698✔
232
      TSDB_CHECK_CODE(code, lino, _end);
9,702!
233
    }
234

235
    // pBlock->info.rows += allNullRow ? 0 : 1;
236
    ++pBlock->info.rows;
4,952✔
237
  } else {
238
    tsdbError("invalid retrieve type:%d, %s", pReader->type, idStr);
×
239
    code = TSDB_CODE_INVALID_PARA;
×
240
    TSDB_CHECK_CODE(code, lino, _end);
×
241
  }
242

243
_end:
×
244
  if (code != TSDB_CODE_SUCCESS) {
10,176!
245
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
246
  }
247
  if (funcTypeBlockArray != NULL) {
10,174✔
248
    taosArrayDestroy(funcTypeBlockArray);
5,223✔
249
  }
250
  return code;
10,176✔
251
}
252

253
static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* idstr) {
10,427✔
254
  int32_t code = TSDB_CODE_SUCCESS;
10,427✔
255
  int32_t lino = 0;
10,427✔
256
  int32_t numOfTables = 0;
10,427✔
257

258
  TSDB_CHECK_NULL(p, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,427!
259

260
  numOfTables = p->numOfTables;
10,427✔
261

262
  if (suid != 0) {
10,427✔
263
    code = metaGetTbTSchemaNotNull(p->pVnode->pMeta, suid, -1, 1, &p->pSchema);
9,255✔
264
    if (TSDB_CODE_SUCCESS != code) {
9,264!
265
      tsdbWarn("stable:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", suid, idstr);
×
266
      if (code == TSDB_CODE_NOT_FOUND) {
×
267
        code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
268
      }
269
      TSDB_CHECK_CODE(code, lino, _end);
×
270
    }
271
  } else {
272
    for (int32_t i = 0; i < numOfTables; ++i) {
1,172!
273
      uint64_t uid = p->pTableList[i].uid;
1,179✔
274
      code = metaGetTbTSchemaMaybeNull(p->pVnode->pMeta, uid, -1, 1, &p->pSchema);
1,179✔
275
      TSDB_CHECK_CODE(code, lino, _end);
1,179!
276
      if (p->pSchema != NULL) {
1,179!
277
        break;
1,179✔
278
      }
279

280
      tsdbWarn("table:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", uid, idstr);
×
281
    }
282

283
    // all queried tables have been dropped already, return immediately.
284
    if (p->pSchema == NULL) {
1,172!
285
      tsdbWarn("all queried tables has been dropped, try next group, %s", idstr);
×
286
      code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
287
      TSDB_CHECK_CODE(code, lino, _end);
×
288
    }
289
  }
290

291
_end:
1,172✔
292
  if (code != TSDB_CODE_SUCCESS) {
10,436!
293
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
294
  }
295
  return code;
10,435✔
296
}
297

298
int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOfTables) {
284✔
299
  int32_t           code = TSDB_CODE_SUCCESS;
284✔
300
  int32_t           lino = 0;
284✔
301
  SCacheRowsReader* pReader = (SCacheRowsReader*)reader;
284✔
302

303
  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
284!
304

305
  pReader->pTableList = pTableIdList;
284✔
306
  pReader->numOfTables = numOfTables;
284✔
307
  pReader->lastTs = INT64_MIN;
284✔
308
  destroySttBlockReader(pReader->pLDataIterArray, NULL);
284✔
309
  pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
284✔
310
  TSDB_CHECK_NULL(pReader->pLDataIterArray, code, lino, _end, terrno);
284!
311

312
_end:
284✔
313
  if (code != TSDB_CODE_SUCCESS) {
284!
314
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
315
  }
316
  return code;
284✔
317
}
318

319
int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
10,421✔
320
                                SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr,
321
                                SArray* pFuncTypeList, SColumnInfo* pPkCol, int32_t numOfPks) {
322
  int32_t           code = TSDB_CODE_SUCCESS;
10,421✔
323
  int32_t           lino = 0;
10,421✔
324
  SCacheRowsReader* p = NULL;
10,421✔
325

326
  TSDB_CHECK_NULL(pVnode, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,421!
327
  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
10,421!
328

329
  *pReader = NULL;
10,421✔
330
  p = taosMemoryCalloc(1, sizeof(SCacheRowsReader));
10,421!
331
  TSDB_CHECK_NULL(p, code, lino, _end, terrno);
10,430!
332

333
  p->type = type;
10,430✔
334
  p->pVnode = pVnode;
10,430✔
335
  p->pTsdb = p->pVnode->pTsdb;
10,430✔
336
  p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = INT64_MAX};
10,430✔
337
  p->info.suid = suid;
10,430✔
338
  p->numOfCols = numOfCols;
10,430✔
339
  p->pCidList = pCidList;
10,430✔
340
  p->pSlotIds = pSlotIds;
10,430✔
341
  p->pFuncTypeList = pFuncTypeList;
10,430✔
342

343
  p->rowKey.numOfPKs = numOfPks;
10,430✔
344
  if (numOfPks > 0) {
10,430✔
345
    TSDB_CHECK_NULL(pPkCol, code, lino, _end, TSDB_CODE_INVALID_PARA);
4,884!
346
    p->rowKey.pks[0].type = pPkCol->type;
4,884✔
347
    if (IS_VAR_DATA_TYPE(pPkCol->type)) {
4,884!
348
      p->rowKey.pks[0].pData = taosMemoryCalloc(1, pPkCol->bytes);
1,474!
349
      if (p->rowKey.pks[0].pData == NULL) {
1,471!
350
        taosMemoryFreeClear(p);
×
351
        code = terrno;
×
352
        TSDB_CHECK_CODE(code, lino, _end);
×
353
      }
354
    }
355

356
    p->pkColumn = *pPkCol;
4,881✔
357
  }
358

359
  if (numOfTables == 0) {
10,427!
360
    *pReader = p;
×
361
    p = NULL;
×
362
    goto _end;
×
363
  }
364

365
  p->pTableList = pTableIdList;
10,427✔
366
  p->numOfTables = numOfTables;
10,427✔
367

368
  code = setTableSchema(p, suid, idstr);
10,427✔
369
  TSDB_CHECK_CODE(code, lino, _end);
10,435!
370

371
  p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES);
10,435!
372
  TSDB_CHECK_NULL(p->transferBuf, code, lino, _end, terrno);
10,436!
373

374
  for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
248,145✔
375
    if (IS_VAR_DATA_TYPE(p->pSchema->columns[i].type)) {
237,710!
376
      p->transferBuf[i] = taosMemoryMalloc(p->pSchema->columns[i].bytes);
34,554!
377
      TSDB_CHECK_NULL(p->transferBuf[i], code, lino, _end, terrno);
34,553!
378
    }
379
  }
380

381
  if (idstr != NULL) {
10,435!
382
    p->idstr = taosStrdup(idstr);
10,442!
383
    TSDB_CHECK_NULL(p->idstr, code, lino, _end, terrno);
10,440!
384
  }
385
  code = taosThreadMutexInit(&p->readerMutex, NULL);
10,433✔
386
  TSDB_CHECK_CODE(code, lino, _end);
10,435!
387

388
  p->lastTs = INT64_MIN;
10,435✔
389

390
  *pReader = p;
10,435✔
391
  p = NULL;
10,435✔
392

393
_end:
10,435✔
394
  if (code != TSDB_CODE_SUCCESS) {
10,435!
395
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
396
    *pReader = NULL;
×
397
  }
398
  if (p != NULL) {
10,435!
399
    tsdbCacherowsReaderClose(p);
×
400
  }
401
  return code;
10,437✔
402
}
403

404
void tsdbCacherowsReaderClose(void* pReader) {
10,439✔
405
  SCacheRowsReader* p = pReader;
10,439✔
406
  if (p == NULL) {
10,439!
407
    return;
×
408
  }
409

410
  if (p->pSchema != NULL && p->transferBuf != NULL) {
10,439!
411
    for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
247,967✔
412
      taosMemoryFreeClear(p->transferBuf[i]);
237,517!
413
    }
414

415
    taosMemoryFree(p->transferBuf);
10,450!
416
    taosMemoryFree(p->pSchema);
10,445!
417
  }
418

419
  taosMemoryFree(p->pCurrSchema);
10,438!
420

421
  if (p->rowKey.numOfPKs > 0) {
10,445✔
422
    for (int32_t i = 0; i < p->rowKey.numOfPKs; i++) {
9,770✔
423
      if (IS_VAR_DATA_TYPE(p->rowKey.pks[i].type)) {
4,885!
424
        taosMemoryFree(p->rowKey.pks[i].pData);
1,475!
425
      }
426
    }
427
  }
428

429
  if (p->pLDataIterArray) {
10,445✔
430
    destroySttBlockReader(p->pLDataIterArray, NULL);
263✔
431
    p->pLDataIterArray = NULL;
263✔
432
  }
433

434
  if (p->pFileReader) {
10,445!
435
    tsdbDataFileReaderClose(&p->pFileReader);
×
436
    p->pFileReader = NULL;
×
437
  }
438

439
  taosMemoryFree((void*)p->idstr);
10,445✔
440
  (void)taosThreadMutexDestroy(&p->readerMutex);
10,444✔
441

442
  if (p->pTableMap) {
10,438✔
443
    void*   pe = NULL;
7,953✔
444
    int32_t iter = 0;
7,953✔
445
    while ((pe = tSimpleHashIterate(p->pTableMap, pe, &iter)) != NULL) {
21,407✔
446
      STableLoadInfo* pInfo = *(STableLoadInfo**)pe;
13,450✔
447
      taosArrayDestroy(pInfo->pTombData);
13,450✔
448
      pInfo->pTombData = NULL;
13,454✔
449
    }
450

451
    tSimpleHashCleanup(p->pTableMap);
7,955✔
452
  }
453
  if (p->uidList) {
10,444✔
454
    taosMemoryFree(p->uidList);
149!
455
  }
456

457
  taosMemoryFree(pReader);
10,444!
458
}
459

460
static int32_t tsdbCacheQueryReseek(void* pQHandle) {
×
461
  int32_t           code = 0;
×
462
  SCacheRowsReader* pReader = pQHandle;
×
463

464
  code = taosThreadMutexTryLock(&pReader->readerMutex);
×
465
  if (code == 0) {
×
466
    // pause current reader's state if not paused, save ts & version for resuming
467
    // just wait for the big all tables' snapshot untaking for now
468

469
    code = TSDB_CODE_VND_QUERY_BUSY;
×
470
    (void)taosThreadMutexUnlock(&pReader->readerMutex);
×
471

472
    return code;
×
473
  } else if (code == EBUSY) {
×
474
    return TSDB_CODE_VND_QUERY_BUSY;
×
475
  } else {
476
    return -1;
×
477
  }
478
}
479

480
int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32_t* slotIds, const int32_t* dstSlotIds,
15,916✔
481
                              SArray* pTableUidList, bool* pGotAll) {
482
  int32_t           code = TSDB_CODE_SUCCESS;
15,916✔
483
  int32_t           lino = 0;
15,916✔
484
  bool              hasRes = false;
15,916✔
485
  SArray*           pRow = NULL;
15,916✔
486
  void**            pRes = NULL;
15,916✔
487
  SCacheRowsReader* pr = NULL;
15,916✔
488
  int32_t           pkBufLen = 0;
15,916✔
489

490
  TSDB_CHECK_NULL(pReader, code, lino, _end, TSDB_CODE_INVALID_PARA);
15,916!
491
  TSDB_CHECK_NULL(pResBlock, code, lino, _end, TSDB_CODE_INVALID_PARA);
15,916!
492

493
  pr = pReader;
15,916✔
494

495
  pr->pReadSnap = NULL;
15,916✔
496
  pRow = taosArrayInit(TARRAY_SIZE(pr->pCidList), sizeof(SLastCol));
15,916✔
497
  TSDB_CHECK_NULL(pRow, code, lino, _end, terrno);
15,924!
498

499
  pRes = taosMemoryCalloc(pr->numOfCols, POINTER_BYTES);
15,924!
500
  TSDB_CHECK_NULL(pRes, code, lino, _end, terrno);
15,933!
501

502
  pkBufLen = (pr->rowKey.numOfPKs > 0) ? pr->pkColumn.bytes : 0;
15,933✔
503
  for (int32_t j = 0; j < pr->numOfCols; ++j) {
44,070✔
504
    int32_t bytes = (slotIds[j] == -1) ? 1 : pr->pSchema->columns[slotIds[j]].bytes;
28,144✔
505

506
    pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + bytes + pkBufLen + VARSTR_HEADER_SIZE);
28,144!
507
    TSDB_CHECK_NULL(pRes[j], code, lino, _end, terrno);
28,137!
508

509
    SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);
28,137✔
510
    p->ts = INT64_MIN;
28,137✔
511
  }
512

513
  (void)taosThreadMutexLock(&pr->readerMutex);
15,926✔
514
  code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap, pr->idstr);
15,931✔
515
  TSDB_CHECK_CODE(code, lino, _end);
15,928!
516

517
  int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
15,928✔
518

519
  STableKeyInfo* pTableList = pr->pTableList;
15,928✔
520

521
  // retrieve the only one last row of all tables in the uid list.
522
  if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
15,928✔
523
    SArray* pLastCols = taosArrayInit(pr->numOfCols, sizeof(SLastCol));
4,346✔
524
    TSDB_CHECK_NULL(pLastCols, code, lino, _end, terrno);
4,341!
525

526
    for (int32_t i = 0; i < pr->numOfCols; ++i) {
11,363✔
527
      int32_t slotId = slotIds[i];
7,016✔
528
      if (slotId == -1) {
7,016✔
529
        SLastCol p = {.rowKey.ts = INT64_MIN, .colVal.value.type = TSDB_DATA_TYPE_BOOL, .colVal.flag = CV_FLAG_NULL};
268✔
530
        void*    px = taosArrayPush(pLastCols, &p);
268✔
531
        TSDB_CHECK_NULL(px, code, lino, _end, terrno);
268!
532
        continue;
268✔
533
      }
534
      struct STColumn* pCol = &pr->pSchema->columns[slotId];
6,748✔
535
      SLastCol         p = {.rowKey.ts = INT64_MIN, .colVal.value.type = pCol->type, .colVal.flag = CV_FLAG_NULL};
6,748✔
536

537
      if (pr->rowKey.numOfPKs > 0) {
6,748✔
538
        p.rowKey.numOfPKs = pr->rowKey.numOfPKs;
2,947✔
539
        for (int32_t j = 0; j < pr->rowKey.numOfPKs; j++) {
5,894✔
540
          p.rowKey.pks[j].type = pr->pkColumn.type;
2,948✔
541
          if (IS_VAR_DATA_TYPE(pr->pkColumn.type)) {
2,948!
542
            p.rowKey.pks[j].pData = taosMemoryCalloc(1, pr->pkColumn.bytes);
900!
543
            TSDB_CHECK_NULL(p.rowKey.pks[j].pData, code, lino, _end, terrno);
899!
544
          }
545
        }
546
      }
547

548
      if (IS_VAR_DATA_TYPE(pCol->type) || pCol->type == TSDB_DATA_TYPE_DECIMAL) {
6,747!
549
        p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char));
1,245!
550
        TSDB_CHECK_NULL(p.colVal.value.pData, code, lino, _end, terrno);
1,253!
551
      }
552

553
      void* px = taosArrayPush(pLastCols, &p);
6,753✔
554
      TSDB_CHECK_NULL(px, code, lino, _end, terrno);
6,753!
555
    }
556

557
    int64_t st = taosGetTimestampUs();
4,344✔
558
    int64_t totalLastTs = INT64_MAX;
4,344✔
559
    for (int32_t i = 0; i < pr->numOfTables; ++i) {
17,650✔
560
      tb_uid_t uid = pTableList[i].uid;
13,310✔
561

562
      code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);
13,310✔
563
      if (code == -1) {  // fix the invalid return code
13,307!
564
        code = 0;
×
565
      }
566
      TSDB_CHECK_CODE(code, lino, _end);
13,307!
567

568
      if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {
13,307!
569
        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
3,381✔
570
        continue;
3,383✔
571
      }
572

573
      {
574
        bool    hasNotNullRow = true;
9,926✔
575
        int64_t singleTableLastTs = INT64_MAX;
9,926✔
576
        for (int32_t k = 0; k < pr->numOfCols; ++k) {
25,515✔
577
          if (slotIds[k] == -1) continue;
15,591✔
578
          SLastCol* p = taosArrayGet(pLastCols, k);
15,415✔
579
          if (p == NULL) {
15,413!
580
            return TSDB_CODE_INVALID_PARA;
×
581
          }
582

583
          SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, k);
15,413✔
584
          if (pColVal == NULL) {
15,411!
585
            return TSDB_CODE_INVALID_PARA;
×
586
          }
587

588
          if (tRowKeyCompare(&pColVal->rowKey, &p->rowKey) > 0) {
15,411✔
589
            if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
7,020✔
590
              if (!COL_VAL_IS_VALUE(&p->colVal)) {
268✔
591
                hasNotNullRow = false;
226✔
592
              }
593
              // For all of cols is null, the last null col of last table will be save
594
              if (i != pr->numOfTables - 1 || k != pr->numOfCols - 1 || hasRes) {
268!
595
                continue;
268✔
596
              }
597
            }
598

599
            hasRes = true;
6,752✔
600
            p->rowKey.ts = pColVal->rowKey.ts;
6,752✔
601
            for (int32_t j = 0; j < p->rowKey.numOfPKs; j++) {
9,698✔
602
              if (IS_VAR_DATA_TYPE(p->rowKey.pks[j].type)) {
2,943!
603
                memcpy(p->rowKey.pks[j].pData, pColVal->rowKey.pks[j].pData, pColVal->rowKey.pks[j].nData);
898✔
604
                p->rowKey.pks[j].nData = pColVal->rowKey.pks[j].nData;
898✔
605
              } else {
606
                valueCloneDatum(p->rowKey.pks + j, pColVal->rowKey.pks + j, p->rowKey.pks[j].type);
2,045✔
607
              }
608
            }
609

610
            if (k == 0) {
6,755✔
611
              if (TARRAY_SIZE(pTableUidList) == 0) {
4,050✔
612
                void* px = taosArrayPush(pTableUidList, &uid);
3,680✔
613
                TSDB_CHECK_NULL(px, code, lino, _end, terrno);
3,680!
614
              } else {
615
                taosArraySet(pTableUidList, 0, &uid);
367✔
616
              }
617
            }
618

619
            if (pColVal->rowKey.ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
6,753✔
620
              singleTableLastTs = pColVal->rowKey.ts;
2,324✔
621
            }
622

623
            if (p->colVal.value.type != pColVal->colVal.value.type) {
6,753!
624
              // check for type/cid mismatch
625
              tsdbError("last cache type mismatch, uid:%" PRIu64
×
626
                        ", schema-type:%d, slotId:%d, cache-type:%d, cache-col:%d",
627
                        uid, p->colVal.value.type, slotIds[k], pColVal->colVal.value.type, pColVal->colVal.cid);
628
              taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
×
629
              code = TSDB_CODE_INVALID_PARA;
×
630
              goto _end;
×
631
            }
632

633
            if (!IS_VAR_DATA_TYPE(pColVal->colVal.value.type) && pColVal->colVal.value.type != TSDB_DATA_TYPE_DECIMAL) {
6,753!
634
              p->colVal = pColVal->colVal;
5,651✔
635
            } else {
636
              if (COL_VAL_IS_VALUE(&pColVal->colVal)) {
1,102✔
637
                memcpy(p->colVal.value.pData, pColVal->colVal.value.pData, pColVal->colVal.value.nData);
1,041✔
638
              }
639

640
              p->colVal.value.nData = pColVal->colVal.value.nData;
1,102✔
641
              p->colVal.value.type = pColVal->colVal.value.type;
1,102✔
642
              p->colVal.flag = pColVal->colVal.flag;
1,102✔
643
              p->colVal.cid = pColVal->colVal.cid;
1,102✔
644
            }
645
          }
646
        }
647

648
        if (hasNotNullRow) {
9,924✔
649
          if (INT64_MAX == totalLastTs || (INT64_MAX != singleTableLastTs && totalLastTs < singleTableLastTs)) {
9,887✔
650
            totalLastTs = singleTableLastTs;
6,597✔
651
          }
652
          double cost = (taosGetTimestampUs() - st) / 1000.0;
9,888✔
653
          if (cost > tsCacheLazyLoadThreshold) {
9,888✔
654
            // pr->lastTs = totalLastTs;
655
          }
656
        }
657
      }
658

659
      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
9,925✔
660
    }
661

662
    if (hasRes) {
4,340✔
663
      code = saveOneRow(pLastCols, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
3,688✔
664
      TSDB_CHECK_CODE(code, lino, _end);
3,686!
665
    }
666

667
    taosArrayDestroyEx(pLastCols, tsdbCacheFreeSLastColItem);
4,338✔
668
  } else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {
11,582✔
669
    int32_t i = pr->tableIndex;
11,580✔
670
    for (; i < pr->numOfTables; ++i) {
14,846✔
671
      tb_uid_t uid = pTableList[i].uid;
8,465✔
672

673
      if ((code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype)) != 0) {
8,465!
674
        if (code == -1) {  // fix the invalid return code
×
675
          code = 0;
×
676
        }
677
        TSDB_CHECK_CODE(code, lino, _end);
×
678
      }
679

680
      if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {
8,470✔
681
        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
1,978✔
682
        continue;
1,978✔
683
      }
684

685
      code = saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
6,492✔
686
      TSDB_CHECK_CODE(code, lino, _end);
6,492!
687

688
      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
6,492✔
689

690
      void* px = taosArrayPush(pTableUidList, &uid);
6,490✔
691
      TSDB_CHECK_NULL(px, code, lino, _end, terrno);
6,490!
692

693
      ++pr->tableIndex;
6,490✔
694
      if (pResBlock->info.rows >= pResBlock->info.capacity) {
6,490✔
695
        break;
5,202✔
696
      }
697
    }
698

699
    if (pGotAll && i == pr->numOfTables) {
11,583!
700
      *pGotAll = true;
6,382✔
701
    }
702
  } else {
703
    code = TSDB_CODE_INVALID_PARA;
2✔
704
    TSDB_CHECK_CODE(code, lino, _end);
2!
705
  }
706

707
_end:
2✔
708
  tsdbUntakeReadSnap2((STsdbReader*)pr, pr->pReadSnap, true);
15,932✔
709
  pr->pReadSnap = NULL;
15,930✔
710

711
  if (pr->pCurFileSet) {
15,930✔
712
    pr->pCurFileSet = NULL;
147✔
713
  }
714

715
  (void)taosThreadMutexUnlock(&pr->readerMutex);
15,930✔
716

717
  if (pRes != NULL) {
15,933!
718
    for (int32_t j = 0; j < pr->numOfCols; ++j) {
44,093✔
719
      taosMemoryFree(pRes[j]);
28,159!
720
    }
721
  }
722

723
  taosMemoryFree(pRes);
15,933!
724
  taosArrayDestroy(pRow);
15,933✔
725

726
  if (code != TSDB_CODE_SUCCESS) {
15,932!
727
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
728
  }
729
  return code;
15,932✔
730
}
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