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

taosdata / TDengine / #3524

08 Nov 2024 04:27AM UTC coverage: 60.898% (+5.0%) from 55.861%
#3524

push

travis-ci

web-flow
Merge pull request #28647 from taosdata/fix/3.0/TD-32519_drop_ctb

fix TD-32519 drop child table with tsma caused crash

118687 of 248552 branches covered (47.75%)

Branch coverage included in aggregate %.

286 of 337 new or added lines in 18 files covered. (84.87%)

9647 existing lines in 190 files now uncovered.

199106 of 273291 relevant lines covered (72.85%)

15236719.35 hits per line

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

76.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) {
32✔
28
  char* buf = taosMemoryCalloc(1, pCol->info.bytes);
32✔
29
  if (buf == NULL) {
32!
30
    return terrno;
×
31
  }
32

33
  SFirstLastRes* pRes = (SFirstLastRes*)((char*)buf + VARSTR_HEADER_SIZE);
32✔
34
  pRes->bytes = 0;
32✔
35
  pRes->hasResult = true;
32✔
36
  pRes->isNull = true;
32✔
37
  varDataSetLen(buf, pCol->info.bytes - VARSTR_HEADER_SIZE);
32✔
38
  int32_t code = colDataSetVal(pCol, row, buf, false);
32✔
39
  taosMemoryFree(buf);
32✔
40

41
  return code;
32✔
42
}
43

44
static int32_t saveOneRowForLastRaw(SLastCol* pColVal, SCacheRowsReader* pReader, const int32_t slotId,
9,847✔
45
                                 SColumnInfoData* pColInfoData, int32_t numOfRows) {
46
  SColVal* pVal = &pColVal->colVal;
9,847✔
47
  int32_t code = 0;
9,847✔
48

49
  // allNullRow = false;
50
  if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {
9,847!
51
    if (!COL_VAL_IS_VALUE(&pColVal->colVal)) {
1,016✔
52
      colDataSetNULL(pColInfoData, numOfRows);
124✔
53
    } else {
54
      varDataSetLen(pReader->transferBuf[slotId], pVal->value.nData);
892✔
55

56
      memcpy(varDataVal(pReader->transferBuf[slotId]), pVal->value.pData, pVal->value.nData);
892✔
57
      code = colDataSetVal(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);
892✔
58
    }
59
  } else {
60
    code = colDataSetVal(pColInfoData, numOfRows, (const char*)&pVal->value.val, !COL_VAL_IS_VALUE(pVal));
8,831✔
61
  }
62

63
  return code;
9,847✔
64
}
65

66
static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,
9,833✔
67
                          const int32_t* dstSlotIds, void** pRes, const char* idStr) {
68
  int32_t numOfRows = pBlock->info.rows;
9,833✔
69
  int32_t code = 0;
9,833✔
70

71
  if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {
9,833✔
72
    uint64_t       ts = TSKEY_MIN;
5,370✔
73
    SFirstLastRes* p = NULL;
5,370✔
74
    col_id_t       colId = -1;
5,370✔
75

76
    SArray* funcTypeBlockArray = taosArrayInit(pReader->numOfCols, sizeof(int32_t));
5,370✔
77
    if (funcTypeBlockArray == NULL) {
5,370!
78
      return terrno;
×
79
    }
80

81
    for (int32_t i = 0; i < pReader->numOfCols; ++i) {
13,010✔
82
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
7,638✔
83
      if (pColInfoData == NULL) {
7,638!
84
        return TSDB_CODE_INVALID_PARA;
×
85
      }
86

87
      int32_t funcType = FUNCTION_TYPE_CACHE_LAST;
7,638✔
88
      if (pReader->pFuncTypeList != NULL && taosArrayGetSize(pReader->pFuncTypeList) > i) {
7,638✔
89
        void* pVal = taosArrayGet(pReader->pFuncTypeList, i);
1,571✔
90
        if (pVal == NULL) {
1,571!
91
          return TSDB_CODE_INVALID_PARA;
×
92
        }
93

94
        funcType = *(int32_t*) pVal;
1,571✔
95
        pVal = taosArrayGet(pReader->pFuncTypeList, i);
1,571✔
96
        if (pVal == NULL) {
1,571!
97
          return TSDB_CODE_INVALID_PARA;
×
98
        }
99

100
        void* px = taosArrayInsert(funcTypeBlockArray, dstSlotIds[i], pVal);
1,571✔
101
        if (px == NULL) {
1,572!
102
          return terrno;
×
103
        }
104
      }
105

106
      if (slotIds[i] == -1) {
7,640✔
107
        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
32!
108
          colDataSetNULL(pColInfoData, numOfRows);
×
109
          continue;
×
110
        }
111

112
        code = setFirstLastResColToNull(pColInfoData, numOfRows);
32✔
113
        if (code) {
32!
114
          return code;
×
115
        }
116
        continue;
32✔
117
      }
118

119
      int32_t   slotId = slotIds[i];
7,608✔
120
      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
7,608✔
121
      if (pColVal == NULL) {
7,608!
122
        return TSDB_CODE_INVALID_PARA;
×
123
      }
124

125
      colId = pColVal->colVal.cid;
7,608✔
126
      if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
7,608✔
127
        code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);
826✔
128
        if (code) {
826!
129
          return code;
×
130
        }
131

132
        continue;
826✔
133
      }
134

135
      p = (SFirstLastRes*)varDataVal(pRes[i]);
6,782✔
136

137
      p->ts = pColVal->rowKey.ts;
6,782✔
138
      ts = p->ts;
6,782✔
139
      p->isNull = !COL_VAL_IS_VALUE(&pColVal->colVal);
6,782✔
140
      // allNullRow = p->isNull & allNullRow;
141
      if (!p->isNull) {
6,782✔
142
        if (IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {
6,770!
143
          varDataSetLen(p->buf, pColVal->colVal.value.nData);
1,797✔
144

145
          memcpy(varDataVal(p->buf), pColVal->colVal.value.pData, pColVal->colVal.value.nData);
1,797✔
146
          p->bytes = pColVal->colVal.value.nData + VARSTR_HEADER_SIZE;  // binary needs to plus the header size
1,797✔
147
        } else {
148
          memcpy(p->buf, &pColVal->colVal.value.val, pReader->pSchema->columns[slotId].bytes);
4,973✔
149
          p->bytes = pReader->pSchema->columns[slotId].bytes;
4,973✔
150
        }
151
      }
152

153
      // pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to subtract it
154
      p->hasResult = true;
6,782✔
155
      varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);
6,782✔
156
      code = colDataSetVal(pColInfoData, numOfRows, (const char*)pRes[i], false);
6,782✔
157
      if (code) {
6,782!
158
        return code;
×
159
      }
160
    }
161

162
    for (int32_t idx = 0; idx < taosArrayGetSize(pBlock->pDataBlock); ++idx) {
22,246✔
163
      SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, idx);
16,875✔
164
      if (pCol == NULL) {
16,874!
165
        return TSDB_CODE_INVALID_PARA;
×
166
      }
167

168
      if (idx < funcTypeBlockArray->size) {
16,874✔
169
        void* pVal = taosArrayGet(funcTypeBlockArray, idx);
1,573✔
170
        if (pVal == NULL) {
1,573!
171
          return TSDB_CODE_INVALID_PARA;
×
172
        }
173

174
        int32_t funcType = *(int32_t*)pVal;
1,573✔
175
        if (FUNCTION_TYPE_CACHE_LAST_ROW == funcType) {
1,573✔
176
          continue;
826✔
177
        }
178
      }
179

180
      if (pCol->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID && pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
16,048✔
181
        if (ts == TSKEY_MIN) {
6,157✔
182
          colDataSetNULL(pCol, numOfRows);
32!
183
        } else {
184
          code = colDataSetVal(pCol, numOfRows, (const char*)&ts, false);
6,125✔
185
          if (code) {
6,125!
186
            return code;
×
187
          }
188
        }
189
        continue;
6,157✔
190
      } else if (pReader->numOfCols == 1 && idx != dstSlotIds[0] && (pCol->info.colId == colId || colId == -1)) {
9,891✔
191
        if (p && !p->isNull) {
1,176!
192
          code = colDataSetVal(pCol, numOfRows, p->buf, false);
1,144✔
193
          if (code) {
1,144!
194
            return code;
×
195
          }
196
        } else {
197
          colDataSetNULL(pCol, numOfRows);
32!
198
        }
199
      }
200
    }
201

202
    // pBlock->info.rows += allNullRow ? 0 : 1;
203
    ++pBlock->info.rows;
5,368✔
204
    taosArrayDestroy(funcTypeBlockArray);
5,368✔
205
  } else if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST_ROW)) {
4,463✔
206
    for (int32_t i = 0; i < pReader->numOfCols; ++i) {
13,832✔
207
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
9,370✔
208
      if (pColInfoData == NULL) {
9,371!
209
        return TSDB_CODE_INVALID_PARA;
×
210
      }
211

212
      int32_t slotId = slotIds[i];
9,371✔
213
      if (slotId == -1) {
9,371✔
214
        colDataSetNULL(pColInfoData, numOfRows);
349!
215
        continue;
349✔
216
      }
217

218
      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
9,022✔
219
      if (pColVal == NULL) {
9,019!
220
        return TSDB_CODE_INVALID_PARA;
×
221
      }
222

223
      code = saveOneRowForLastRaw(pColVal, pReader, slotId, pColInfoData, numOfRows);
9,019✔
224
      if (code) {
9,021!
225
        return code;
×
226
      }
227
    }
228

229
    // pBlock->info.rows += allNullRow ? 0 : 1;
230
    ++pBlock->info.rows;
4,462✔
231
  } else {
232
    tsdbError("invalid retrieve type:%d, %s", pReader->type, idStr);
1!
233
    return TSDB_CODE_INVALID_PARA;
×
234
  }
235

236
  return code;
9,831✔
237
}
238

239
static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* idstr) {
10,016✔
240
  int32_t numOfTables = p->numOfTables;
10,016✔
241
  int32_t  code = TSDB_CODE_SUCCESS;
10,016✔
242

243
  if (suid != 0) {
10,016✔
244
    code = metaGetTbTSchemaNotNull(p->pVnode->pMeta, suid, -1, 1, &p->pSchema);
8,860✔
245
    if (TSDB_CODE_SUCCESS != code) {
8,868✔
246
      tsdbWarn("stable:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", suid, idstr);
2!
247
      if(code == TSDB_CODE_NOT_FOUND) {
×
248
        return TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
249
      }  else {
250
        return code;
×
251
      }
252
    }
253
  } else {
254
    for (int32_t i = 0; i < numOfTables; ++i) {
1,156!
255
      uint64_t uid = p->pTableList[i].uid;
1,160✔
256
      code = metaGetTbTSchemaMaybeNull(p->pVnode->pMeta, uid, -1, 1, &p->pSchema);
1,160✔
257
      if(code != TSDB_CODE_SUCCESS) {
1,160!
258
        return code;
×
259
      }
260
      if (p->pSchema != NULL) {
1,160!
261
        break;
1,160✔
262
      }
263

264
      tsdbWarn("table:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", uid, idstr);
×
265
    }
266

267
    // all queried tables have been dropped already, return immediately.
268
    if (p->pSchema == NULL) {
1,156!
269
      tsdbWarn("all queried tables has been dropped, try next group, %s", idstr);
×
270
      return TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
271
    }
272
  }
273

274
  return TSDB_CODE_SUCCESS;
10,022✔
275
}
276

277
int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOfTables) {
294✔
278
  SCacheRowsReader* pReader = (SCacheRowsReader*)reader;
294✔
279

280
  pReader->pTableList = pTableIdList;
294✔
281
  pReader->numOfTables = numOfTables;
294✔
282
  pReader->lastTs = INT64_MIN;
294✔
283
  destroySttBlockReader(pReader->pLDataIterArray, NULL);
294✔
284
  pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
294✔
285

286
  return (pReader->pLDataIterArray != NULL) ? TSDB_CODE_SUCCESS : terrno;
294!
287
}
288

289
int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
10,017✔
290
                                SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr,
291
                                SArray* pFuncTypeList, SColumnInfo* pPkCol, int32_t numOfPks) {
292
  *pReader = NULL;
10,017✔
293
  SCacheRowsReader* p = taosMemoryCalloc(1, sizeof(SCacheRowsReader));
10,017✔
294
  if (p == NULL) {
10,016!
295
    return terrno;
×
296
  }
297

298
  p->type = type;
10,016✔
299
  p->pVnode = pVnode;
10,016✔
300
  p->pTsdb = p->pVnode->pTsdb;
10,016✔
301
  p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = INT64_MAX};
10,016✔
302
  p->info.suid = suid;
10,016✔
303
  p->numOfCols = numOfCols;
10,016✔
304
  p->pCidList = pCidList;
10,016✔
305
  p->pSlotIds = pSlotIds;
10,016✔
306
  p->pFuncTypeList = pFuncTypeList;
10,016✔
307

308
  p->rowKey.numOfPKs = numOfPks;
10,016✔
309
  if (numOfPks > 0) {
10,016✔
310
    p->rowKey.pks[0].type = pPkCol->type;
4,289✔
311
    if (IS_VAR_DATA_TYPE(pPkCol->type)) {
4,289!
312
      p->rowKey.pks[0].pData = taosMemoryCalloc(1, pPkCol->bytes);
1,298✔
313
      if (p->rowKey.pks[0].pData == NULL) {
1,298!
314
        taosMemoryFree(p);
×
315
        return terrno;
×
316
      }
317
    }
318

319
    p->pkColumn = *pPkCol;
4,289✔
320
  }
321

322
  if (numOfTables == 0) {
10,016!
323
    *pReader = p;
×
324
    return TSDB_CODE_SUCCESS;
×
325
  }
326

327
  p->pTableList = pTableIdList;
10,016✔
328
  p->numOfTables = numOfTables;
10,016✔
329

330
  int32_t code = setTableSchema(p, suid, idstr);
10,016✔
331
  if (code != TSDB_CODE_SUCCESS) {
10,027!
332
    tsdbCacherowsReaderClose(p);
×
333
    return code;
×
334
  }
335

336
  p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES);
10,027✔
337
  if (p->transferBuf == NULL) {
10,020✔
338
    tsdbCacherowsReaderClose(p);
1✔
339
    return terrno;
×
340
  }
341

342
  for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
223,032✔
343
    if (IS_VAR_DATA_TYPE(p->pSchema->columns[i].type)) {
213,028!
344
      p->transferBuf[i] = taosMemoryMalloc(p->pSchema->columns[i].bytes);
34,147✔
345
      if (p->transferBuf[i] == NULL) {
34,132!
UNCOV
346
        tsdbCacherowsReaderClose(p);
×
UNCOV
347
        return terrno;
×
348
      }
349
    }
350
  }
351

352
  p->idstr = taosStrdup(idstr);
10,004✔
353
  if (idstr != NULL && p->idstr == NULL) {
10,023!
354
    tsdbCacherowsReaderClose(p);
×
355
    return terrno;
×
356
  }
357
  code = taosThreadMutexInit(&p->readerMutex, NULL);
10,023✔
358
  if (code) {
10,017!
359
    tsdbCacherowsReaderClose(p);
×
360
    return code;
×
361
  }
362

363
  p->lastTs = INT64_MIN;
10,021✔
364

365
  *pReader = p;
10,021✔
366
  return code;
10,021✔
367
}
368

369
void tsdbCacherowsReaderClose(void* pReader) {
10,027✔
370
  SCacheRowsReader* p = pReader;
10,027✔
371
  if (p == NULL) {
10,027!
UNCOV
372
    return;
×
373
  }
374

375
  if (p->pSchema != NULL && p->transferBuf != NULL) {
10,027!
376
    for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
223,300✔
377
      taosMemoryFreeClear(p->transferBuf[i]);
213,279✔
378
    }
379

380
    taosMemoryFree(p->transferBuf);
10,021✔
381
    taosMemoryFree(p->pSchema);
10,029✔
382
  }
383

384
  taosMemoryFree(p->pCurrSchema);
10,027✔
385

386
  if (p->rowKey.numOfPKs > 0) {
10,026✔
387
    for (int32_t i = 0; i < p->rowKey.numOfPKs; i++) {
8,577✔
388
      if (IS_VAR_DATA_TYPE(p->rowKey.pks[i].type)) {
4,287!
389
        taosMemoryFree(p->rowKey.pks[i].pData);
1,296✔
390
      }
391
    }
392
  }
393

394
  if (p->pLDataIterArray) {
10,028✔
395
    destroySttBlockReader(p->pLDataIterArray, NULL);
288✔
396
  }
397

398
  if (p->pFileReader) {
10,029✔
399
    tsdbDataFileReaderClose(&p->pFileReader);
3✔
400
    p->pFileReader = NULL;
3✔
401
  }
402

403
  taosMemoryFree((void*)p->idstr);
10,029✔
404
  (void) taosThreadMutexDestroy(&p->readerMutex);
10,029✔
405

406
  if (p->pTableMap) {
10,028✔
407
    void*   pe = NULL;
329✔
408
    int32_t iter = 0;
329✔
409
    while ((pe = tSimpleHashIterate(p->pTableMap, pe, &iter)) != NULL) {
1,137✔
410
      STableLoadInfo* pInfo = *(STableLoadInfo**)pe;
807✔
411
      taosArrayDestroy(pInfo->pTombData);
807✔
412
      pInfo->pTombData = NULL;
808✔
413
    }
414

415
    tSimpleHashCleanup(p->pTableMap);
330✔
416
  }
417
  if (p->uidList) {
10,029✔
418
    taosMemoryFree(p->uidList);
172✔
419
  }
420

421
  taosMemoryFree(pReader);
10,029✔
422
}
423

424
static int32_t tsdbCacheQueryReseek(void* pQHandle) {
×
425
  int32_t           code = 0;
×
426
  SCacheRowsReader* pReader = pQHandle;
×
427

428
  code = taosThreadMutexTryLock(&pReader->readerMutex);
×
429
  if (code == 0) {
×
430
    // pause current reader's state if not paused, save ts & version for resuming
431
    // just wait for the big all tables' snapshot untaking for now
432

433
    code = TSDB_CODE_VND_QUERY_BUSY;
×
434
    (void)taosThreadMutexUnlock(&pReader->readerMutex);
×
435

436
    return code;
×
437
  } else if (code == EBUSY) {
×
438
    return TSDB_CODE_VND_QUERY_BUSY;
×
439
  } else {
440
    return -1;
×
441
  }
442
}
443

444
int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32_t* slotIds, const int32_t* dstSlotIds,
14,554✔
445
                              SArray* pTableUidList, bool* pGotAll) {
446
  if (pReader == NULL || pResBlock == NULL) {
14,554!
447
    return TSDB_CODE_INVALID_PARA;
×
448
  }
449

450
  int32_t           code = TSDB_CODE_SUCCESS;
14,559✔
451
  bool              hasRes = false;
14,559✔
452
  SArray*           pRow = NULL;
14,559✔
453
  void**            pRes = NULL;
14,559✔
454
  SCacheRowsReader* pr = pReader;
14,559✔
455
  int32_t           pkBufLen = 0;
14,559✔
456

457
  pr->pReadSnap = NULL;
14,559✔
458
  pRow = taosArrayInit(TARRAY_SIZE(pr->pCidList), sizeof(SLastCol));
14,559✔
459
  if (pRow == NULL) {
14,560!
460
    code = terrno;
×
461
    goto _end;
×
462
  }
463

464
  pRes = taosMemoryCalloc(pr->numOfCols, POINTER_BYTES);
14,560✔
465
  if (pRes == NULL) {
14,571✔
466
    code = terrno;
3✔
467
    goto _end;
×
468
  }
469

470
  pkBufLen = (pr->rowKey.numOfPKs > 0) ? pr->pkColumn.bytes : 0;
14,568✔
471
  for (int32_t j = 0; j < pr->numOfCols; ++j) {
40,059✔
472
    int32_t bytes = (slotIds[j] == -1) ? 1 : pr->pSchema->columns[slotIds[j]].bytes;
25,491✔
473

474
    pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + bytes + pkBufLen + VARSTR_HEADER_SIZE);
25,491✔
475
    if (pRes[j] == NULL) {
25,487!
476
      code = terrno;
×
477
      goto _end;
×
478
    }
479

480
    SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);
25,491✔
481
    p->ts = INT64_MIN;
25,491✔
482
  }
483

484
  (void)taosThreadMutexLock(&pr->readerMutex);
14,568✔
485
  code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap, pr->idstr);
14,570✔
486
  if (code != TSDB_CODE_SUCCESS) {
14,570!
487
    goto _end;
×
488
  }
489

490
  int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
14,570✔
491

492
  STableKeyInfo* pTableList = pr->pTableList;
14,570✔
493

494
  // retrieve the only one last row of all tables in the uid list.
495
  if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
14,570✔
496
    SArray* pLastCols = taosArrayInit(pr->numOfCols, sizeof(SLastCol));
4,684✔
497
    if (pLastCols == NULL) {
4,679!
498
      code = terrno;
×
499
      goto _end;
×
500
    }
501

502
    for (int32_t i = 0; i < pr->numOfCols; ++i) {
12,002✔
503
      int32_t slotId = slotIds[i];
7,318✔
504
      if (slotId == -1) {
7,318✔
505
        SLastCol p = {.rowKey.ts = INT64_MIN, .colVal.value.type = TSDB_DATA_TYPE_BOOL, .colVal.flag = CV_FLAG_NULL};
270✔
506
        void*    px = taosArrayPush(pLastCols, &p);
270✔
507
        if (px == NULL) {
270!
508
          code = terrno;
×
509
          goto _end;
×
510
        }
511
        continue;
270✔
512
      }
513
      struct STColumn* pCol = &pr->pSchema->columns[slotId];
7,048✔
514
      SLastCol         p = {.rowKey.ts = INT64_MIN, .colVal.value.type = pCol->type, .colVal.flag = CV_FLAG_NULL};
7,048✔
515

516
      if (pr->rowKey.numOfPKs > 0) {
7,048✔
517
        p.rowKey.numOfPKs = pr->rowKey.numOfPKs;
3,095✔
518
        for (int32_t j = 0; j < pr->rowKey.numOfPKs; j++) {
6,191✔
519
          p.rowKey.pks[j].type = pr->pkColumn.type;
3,096✔
520
          if (IS_VAR_DATA_TYPE(pr->pkColumn.type)) {
3,096!
521

522
            p.rowKey.pks[j].pData = taosMemoryCalloc(1, pr->pkColumn.bytes);
1,110✔
523
            if (p.rowKey.pks[j].pData == NULL) {
1,110!
524
              code = terrno;
×
525
              goto _end;
×
526
            }
527
          }
528
        }
529
      }
530

531
      if (IS_VAR_DATA_TYPE(pCol->type)) {
7,048!
532
        p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char));
1,331✔
533
        if (p.colVal.value.pData == NULL) {
1,336!
534
          code = terrno;
×
535
          goto _end;
×
536
        }
537
      }
538

539
      void* px = taosArrayPush(pLastCols, &p);
7,053✔
540
      if (px == NULL) {
7,053!
541
        code = terrno;
×
542
        goto _end;
×
543
      }
544
    }
545

546
    int64_t st = taosGetTimestampUs();
4,679✔
547
    int64_t totalLastTs = INT64_MAX;
4,679✔
548
    for (int32_t i = 0; i < pr->numOfTables; ++i) {
19,393✔
549
      tb_uid_t uid = pTableList[i].uid;
14,704✔
550

551
      code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);
14,704✔
552
      if (code == -1) {// fix the invalid return code
14,700!
553
        code = 0;
×
554
      } else if (code != 0) {
14,700!
UNCOV
555
        goto _end;
×
556
      }
557

558
      if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {
14,700!
559
        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
3,852✔
560
        continue;
3,865✔
561
      }
562

563
      {
564
        bool    hasNotNullRow = true;
10,848✔
565
        int64_t singleTableLastTs = INT64_MAX;
10,848✔
566
        for (int32_t k = 0; k < pr->numOfCols; ++k) {
27,023✔
567
          if (slotIds[k] == -1) continue;
16,175✔
568
          SLastCol* p = taosArrayGet(pLastCols, k);
16,047✔
569
          if (p == NULL) {
16,045!
570
            return TSDB_CODE_INVALID_PARA;
×
571
          }
572

573
          SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, k);
16,045✔
574
          if (pColVal == NULL) {
16,047!
575
            return TSDB_CODE_INVALID_PARA;
×
576
          }
577

578
          if (tRowKeyCompare(&pColVal->rowKey, &p->rowKey) > 0) {
16,047✔
579
            if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
7,579✔
580
              if (!COL_VAL_IS_VALUE(&p->colVal)) {
252✔
581
                hasNotNullRow = false;
226✔
582
              }
583
              // For all of cols is null, the last null col of last table will be save
584
              if (i != pr->numOfTables - 1 || k != pr->numOfCols - 1 || hasRes) {
252!
585
                continue;
252✔
586
              }
587
            }
588

589
            hasRes = true;
7,327✔
590
            p->rowKey.ts = pColVal->rowKey.ts;
7,327✔
591
            for (int32_t j = 0; j < p->rowKey.numOfPKs; j++) {
10,420✔
592
              if (IS_VAR_DATA_TYPE(p->rowKey.pks[j].type)) {
3,093!
593
                memcpy(p->rowKey.pks[j].pData, pColVal->rowKey.pks[j].pData, pColVal->rowKey.pks[j].nData);
1,108✔
594
                p->rowKey.pks[j].nData = pColVal->rowKey.pks[j].nData;
1,108✔
595
              } else {
596
                p->rowKey.pks[j].val = pColVal->rowKey.pks[j].val;
1,985✔
597
              }
598
            }
599

600
            if (k == 0) {
7,327✔
601
              if (TARRAY_SIZE(pTableUidList) == 0) {
4,533✔
602
                void* px = taosArrayPush(pTableUidList, &uid);
3,862✔
603
                if (px == NULL) {
3,862!
604
                  code = terrno;
×
605
                  goto _end;
×
606
                }
607
              } else {
608
                taosArraySet(pTableUidList, 0, &uid);
670✔
609
              }
610
            }
611

612
            if (pColVal->rowKey.ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
7,327✔
613
              singleTableLastTs = pColVal->rowKey.ts;
2,961✔
614
            }
615

616
            if (p->colVal.value.type != pColVal->colVal.value.type) {
7,327!
617
              // check for type/cid mismatch
UNCOV
618
              tsdbError("last cache type mismatch, uid:%" PRIu64
×
619
                        ", schema-type:%d, slotId:%d, cache-type:%d, cache-col:%d",
620
                        uid, p->colVal.value.type, slotIds[k], pColVal->colVal.value.type, pColVal->colVal.cid);
UNCOV
621
              taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
×
UNCOV
622
              code = TSDB_CODE_INVALID_PARA;
×
UNCOV
623
              goto _end;
×
624
            }
625

626
            if (!IS_VAR_DATA_TYPE(pColVal->colVal.value.type)) {
7,327!
627
              p->colVal = pColVal->colVal;
5,999✔
628
            } else {
629
              if (COL_VAL_IS_VALUE(&pColVal->colVal)) {
1,328✔
630
                memcpy(p->colVal.value.pData, pColVal->colVal.value.pData, pColVal->colVal.value.nData);
1,279✔
631
              }
632

633
              p->colVal.value.nData = pColVal->colVal.value.nData;
1,328✔
634
              p->colVal.value.type = pColVal->colVal.value.type;
1,328✔
635
              p->colVal.flag = pColVal->colVal.flag;
1,328✔
636
              p->colVal.cid = pColVal->colVal.cid;
1,328✔
637
            }
638
          }
639
        }
640

641
        if (hasNotNullRow) {
10,848✔
642
          if (INT64_MAX == totalLastTs || (INT64_MAX != singleTableLastTs && totalLastTs < singleTableLastTs)) {
10,821✔
643
            totalLastTs = singleTableLastTs;
6,710✔
644
          }
645
          double cost = (taosGetTimestampUs() - st) / 1000.0;
10,816✔
646
          if (cost > tsCacheLazyLoadThreshold) {
10,816✔
647
            // pr->lastTs = totalLastTs;
648
          }
649
        }
650
      }
651

652
      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
10,843✔
653
    }
654

655
    if (hasRes) {
4,689✔
656
      code = saveOneRow(pLastCols, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
3,868✔
657
      if (code) {
3,867!
UNCOV
658
        goto _end;
×
659
      }
660
    }
661

662
    taosArrayDestroyEx(pLastCols, tsdbCacheFreeSLastColItem);
4,688✔
663
  } else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {
9,886✔
664
    int32_t i = pr->tableIndex;
9,884✔
665
    for (; i < pr->numOfTables; ++i) {
13,897✔
666
      tb_uid_t uid = pTableList[i].uid;
8,252✔
667

668
      if ((code = tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype)) != 0) {
8,252!
UNCOV
669
        if (code == -1) {// fix the invalid return code
×
UNCOV
670
          code = 0;
×
UNCOV
671
        } else if (code != 0) {
×
UNCOV
672
          goto _end;
×
673
        }
674
      }
675

676
      if (TARRAY_SIZE(pRow) <= 0 || COL_VAL_IS_NONE(&((SLastCol*)TARRAY_DATA(pRow))[0].colVal)) {
8,255✔
677
        taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
2,288✔
678
        continue;
2,295✔
679
      }
680

681
      code = saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
5,967✔
682
      if (code) {
5,966!
UNCOV
683
        goto _end;
×
684
      }
685

686
      taosArrayClearEx(pRow, tsdbCacheFreeSLastColItem);
5,966✔
687

688
      void* px = taosArrayPush(pTableUidList, &uid);
5,968✔
689
      if (px == NULL) {
5,968!
UNCOV
690
        code = terrno;
×
UNCOV
691
        goto _end;
×
692
      }
693

694
      ++pr->tableIndex;
5,968✔
695
      if (pResBlock->info.rows >= pResBlock->info.capacity) {
5,968✔
696
        break;
4,250✔
697
      }
698
    }
699

700
    if (pGotAll && i == pr->numOfTables) {
9,895✔
701
      *pGotAll = true;
5,639✔
702
    }
703
  } else {
704
    code = TSDB_CODE_INVALID_PARA;
2✔
705
  }
706

707
_end:
14,580✔
708
  tsdbUntakeReadSnap2((STsdbReader*)pr, pr->pReadSnap, true);
14,580✔
709
  pr->pReadSnap = NULL;
14,570✔
710

711
  if (pr->pCurFileSet) {
14,570✔
712
    pr->pCurFileSet = NULL;
166✔
713
  }
714

715
  (void)taosThreadMutexUnlock(&pr->readerMutex);
14,570✔
716

717
  if (pRes != NULL) {
14,572!
718
    for (int32_t j = 0; j < pr->numOfCols; ++j) {
40,046✔
719
      taosMemoryFree(pRes[j]);
25,475✔
720
    }
721
  }
722

723
  taosMemoryFree(pRes);
14,571✔
724
  taosArrayDestroy(pRow);
14,570✔
725

726
  return code;
14,568✔
727
}
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