tsdbCacheRead.c 15.5 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
/*
 * 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 "taoserror.h"
#include "tarray.h"
#include "tcommon.h"
#include "tsdb.h"

21
#define HASTYPE(_type, _t) (((_type) & (_t)) == (_t))
22

H
Haojun Liao 已提交
23
static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,
24
                          const int32_t* dstSlotIds, void** pRes, const char* idStr) {
25
  int32_t numOfRows = pBlock->info.rows;
M
Minglei Jin 已提交
26
  bool    allNullRow = true;
27

28
  if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {
29
    for (int32_t i = 0; i < pReader->numOfCols; ++i) {
30
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
31
      SFirstLastRes*   p = (SFirstLastRes*)varDataVal(pRes[i]);
M
Minglei Jin 已提交
32 33
      int32_t          slotId = slotIds[i];
      SLastCol*        pColVal = (SLastCol*)taosArrayGet(pRow, i);
34

35 36 37
      p->ts = pColVal->ts;
      p->isNull = !COL_VAL_IS_VALUE(&pColVal->colVal);
      allNullRow = p->isNull & allNullRow;
38

39 40 41
      if (!p->isNull) {
        if (IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
          varDataSetLen(p->buf, pColVal->colVal.value.nData);
42

43 44 45 46 47
          memcpy(varDataVal(p->buf), pColVal->colVal.value.pData, pColVal->colVal.value.nData);
          p->bytes = pColVal->colVal.value.nData + VARSTR_HEADER_SIZE;  // binary needs to plus the header size
        } else {
          memcpy(p->buf, &pColVal->colVal.value, pReader->pSchema->columns[slotId].bytes);
          p->bytes = pReader->pSchema->columns[slotId].bytes;
48 49
        }
      }
50

51
      // pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to substruct it
52
      p->hasResult = true;
53
      varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);
54
      colDataSetVal(pColInfoData, numOfRows, (const char*)pRes[i], false);
55
    }
56

57
    pBlock->info.rows += allNullRow ? 0 : 1;
H
Haojun Liao 已提交
58
  } else if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST_ROW)) {
59
    for (int32_t i = 0; i < pReader->numOfCols; ++i) {
M
Minglei Jin 已提交
60
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotIds[i]);
61

M
Minglei Jin 已提交
62 63 64
      int32_t   slotId = slotIds[i];
      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, i);
      SColVal*  pVal = &pColVal->colVal;
65

M
Minglei Jin 已提交
66
      allNullRow = false;
M
Minglei Jin 已提交
67 68 69
      if (IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
        if (!COL_VAL_IS_VALUE(&pColVal->colVal)) {
          colDataSetNULL(pColInfoData, numOfRows);
70
        } else {
M
Minglei Jin 已提交
71
          varDataSetLen(pReader->transferBuf[slotId], pVal->value.nData);
72

M
Minglei Jin 已提交
73 74
          memcpy(varDataVal(pReader->transferBuf[slotId]), pVal->value.pData, pVal->value.nData);
          colDataSetVal(pColInfoData, numOfRows, pReader->transferBuf[slotId], false);
75
        }
M
Minglei Jin 已提交
76 77
      } else {
        colDataSetVal(pColInfoData, numOfRows, (const char*)&pVal->value.val, !COL_VAL_IS_VALUE(pVal));
78 79
      }
    }
80

M
Minglei Jin 已提交
81
    pBlock->info.rows += allNullRow ? 0 : 1;
H
Haojun Liao 已提交
82 83 84
  } else {
    tsdbError("invalid retrieve type:%d, %s", pReader->type, idStr);
    return TSDB_CODE_INVALID_PARA;
85
  }
H
Haojun Liao 已提交
86 87

  return TSDB_CODE_SUCCESS;
88 89
}

90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119
static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* idstr) {
  int32_t numOfTables = p->numOfTables;

  if (suid != 0) {
    p->pSchema = metaGetTbTSchema(p->pVnode->pMeta, suid, -1, 1);
    if (p->pSchema == NULL) {
      tsdbWarn("stable:%" PRIu64 " has been dropped, failed to retrieve cached rows, %s", suid, idstr);
      return TSDB_CODE_PAR_TABLE_NOT_EXIST;
    }
  } else {
    for (int32_t i = 0; i < numOfTables; ++i) {
      uint64_t uid = p->pTableList[i].uid;
      p->pSchema = metaGetTbTSchema(p->pVnode->pMeta, uid, -1, 1);
      if (p->pSchema != NULL) {
        break;
      }

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

    // all queried tables have been dropped already, return immediately.
    if (p->pSchema == NULL) {
      tsdbWarn("all queried tables has been dropped, try next group, %s", idstr);
      return TSDB_CODE_PAR_TABLE_NOT_EXIST;
    }
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
120 121 122 123 124 125 126
int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOfTables) {
  SCacheRowsReader* pReader = (SCacheRowsReader*)reader;

  pReader->pTableList = pTableIdList;
  pReader->numOfTables = numOfTables;
  pReader->lastTs = INT64_MIN;

M
Minglei Jin 已提交
127 128 129 130
  int64_t blocks;
  double  elapse;
  pReader->pLDataIterArray = destroySttBlockReader(pReader->pLDataIterArray, &blocks, &elapse);
  pReader->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
D
dapan1121 已提交
131 132 133 134

  return TSDB_CODE_SUCCESS;
}

135 136 137 138 139 140 141 142 143 144 145 146 147 148 149
static int32_t uidComparFunc(const void* p1, const void* p2) {
  uint64_t pu1 = *(uint64_t*)p1;
  uint64_t pu2 = *(uint64_t*)p2;
  if (pu1 == pu2) {
    return 0;
  } else {
    return (pu1 < pu2) ? -1 : 1;
  }
}

static void freeTableInfoFunc(void* param) {
  void** p = (void**)param;
  taosMemoryFreeClear(*p);
}

150
int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
M
Minglei Jin 已提交
151
                                SArray* pCidList, int32_t* pSlotIds, uint64_t suid, void** pReader, const char* idstr) {
152 153
  *pReader = NULL;
  SCacheRowsReader* p = taosMemoryCalloc(1, sizeof(SCacheRowsReader));
154 155 156 157
  if (p == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

158 159
  p->type = type;
  p->pVnode = pVnode;
160
  p->pTsdb = p->pVnode->pTsdb;
161
  p->info.verRange = (SVersionRange){.minVer = 0, .maxVer = UINT64_MAX};
162
  p->info.suid = suid;
163
  p->numOfCols = numOfCols;
164
  p->pCidList = pCidList;
M
Minglei Jin 已提交
165
  p->pSlotIds = pSlotIds;
166

167
  if (numOfTables == 0) {
168 169 170 171
    *pReader = p;
    return TSDB_CODE_SUCCESS;
  }

172
  p->pTableList = pTableIdList;
173
  p->numOfTables = numOfTables;
174

175 176 177 178 179 180 181 182 183 184 185 186 187
  p->pTableMap = tSimpleHashInit(numOfTables, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
  if (p->pTableMap == NULL) {
    tsdbCacherowsReaderClose(p);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  p->uidList = taosMemoryMalloc(numOfTables * sizeof(uint64_t));
  if (p->uidList == NULL) {
    tsdbCacherowsReaderClose(p);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  for (int32_t i = 0; i < numOfTables; ++i) {
    uint64_t uid = p->pTableList[i].uid;
    p->uidList[i] = uid;
188
    STableLoadInfo* pInfo = taosMemoryCalloc(1, sizeof(STableLoadInfo));
189 190 191 192 193 194 195
    tSimpleHashPut(p->pTableMap, &uid, sizeof(uint64_t), &pInfo, POINTER_BYTES);
  }

  tSimpleHashSetFreeFp(p->pTableMap, freeTableInfoFunc);

  taosSort(p->uidList, numOfTables, sizeof(uint64_t), uidComparFunc);

196 197 198 199 200 201
  int32_t code = setTableSchema(p, suid, idstr);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbCacherowsReaderClose(p);
    return code;
  }

202
  p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES);
203
  if (p->transferBuf == NULL) {
H
Haojun Liao 已提交
204
    tsdbCacherowsReaderClose(p);
H
Haojun Liao 已提交
205
    return TSDB_CODE_OUT_OF_MEMORY;
206 207
  }

208
  for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
M
Minglei Jin 已提交
209
    if (IS_VAR_DATA_TYPE(p->pSchema->columns[i].type)) {
210
      p->transferBuf[i] = taosMemoryMalloc(p->pSchema->columns[i].bytes);
211 212 213 214
      if (p->transferBuf[i] == NULL) {
        tsdbCacherowsReaderClose(p);
        return TSDB_CODE_OUT_OF_MEMORY;
      }
215 216
    }
  }
217

218
  SVnodeCfg* pCfg = &((SVnode*)pVnode)->config;
219
  int32_t    numOfStt = pCfg->sttTrigger;
M
Minglei Jin 已提交
220 221
  p->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
  if (p->pLDataIterArray == NULL) {
222 223 224 225
    tsdbCacherowsReaderClose(p);
    return TSDB_CODE_OUT_OF_MEMORY;
  }

226
  p->idstr = taosStrdup(idstr);
227 228
  taosThreadMutexInit(&p->readerMutex, NULL);

229 230
  p->lastTs = INT64_MIN;

231 232 233 234
  *pReader = p;
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
235
void* tsdbCacherowsReaderClose(void* pReader) {
236
  SCacheRowsReader* p = pReader;
H
Haojun Liao 已提交
237 238 239
  if (p == NULL) {
    return NULL;
  }
240

241 242 243 244 245 246 247
  if (p->pSchema != NULL) {
    for (int32_t i = 0; i < p->pSchema->numOfCols; ++i) {
      taosMemoryFreeClear(p->transferBuf[i]);
    }

    taosMemoryFree(p->transferBuf);
    taosMemoryFree(p->pSchema);
248 249
  }

250 251
  taosMemoryFree(p->pCurrSchema);

M
Minglei Jin 已提交
252 253 254
  int64_t loadBlocks = 0;
  double  elapse = 0;
  destroySttBlockReader(p->pLDataIterArray, &loadBlocks, &elapse);
255

256
  taosMemoryFree((void*)p->idstr);
257 258
  taosThreadMutexDestroy(&p->readerMutex);

259 260 261 262 263 264 265 266 267 268 269 270 271 272
  if (p->pTableMap) {
    void*   pe = NULL;
    int32_t iter = 0;
    while ((pe = tSimpleHashIterate(p->pTableMap, pe, &iter)) != NULL) {
      STableLoadInfo* pInfo = *(STableLoadInfo**)pe;
      pInfo->pTombData = taosArrayDestroy(pInfo->pTombData);
    }

    tSimpleHashCleanup(p->pTableMap);
  }
  if (p->uidList) {
    taosMemoryFree(p->uidList);
  }

273
  taosMemoryFree(pReader);
H
Haojun Liao 已提交
274
  return NULL;
275 276
}

277
static void freeItem(void* pItem) {
278
  SLastCol* pCol = (SLastCol*)pItem;
279
  if (IS_VAR_DATA_TYPE(pCol->colVal.type) && pCol->colVal.value.pData) {
280 281 282 283
    taosMemoryFree(pCol->colVal.value.pData);
  }
}

284
static int32_t tsdbCacheQueryReseek(void* pQHandle) {
285 286 287
  int32_t           code = 0;
  SCacheRowsReader* pReader = pQHandle;

288 289 290 291
  code = taosThreadMutexTryLock(&pReader->readerMutex);
  if (code == 0) {
    // pause current reader's state if not paused, save ts & version for resuming
    // just wait for the big all tables' snapshot untaking for now
292

293
    code = TSDB_CODE_VND_QUERY_BUSY;
294

295 296 297 298 299 300 301 302
    taosThreadMutexUnlock(&pReader->readerMutex);

    return code;
  } else if (code == EBUSY) {
    return TSDB_CODE_VND_QUERY_BUSY;
  } else {
    return -1;
  }
303 304
}

305 306
int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32_t* slotIds, const int32_t* dstSlotIds,
                              SArray* pTableUidList) {
307
  if (pReader == NULL || pResBlock == NULL) {
308 309 310
    return TSDB_CODE_INVALID_PARA;
  }

311
  SCacheRowsReader* pr = pReader;
M
Minglei Jin 已提交
312 313 314 315
  int32_t           code = TSDB_CODE_SUCCESS;
  SArray*           pRow = taosArrayInit(TARRAY_SIZE(pr->pCidList), sizeof(SLastCol));
  bool              hasRes = false;
  SArray*           pLastCols = NULL;
316

317
  void** pRes = taosMemoryCalloc(pr->numOfCols, POINTER_BYTES);
318 319 320 321 322
  if (pRes == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _end;
  }

323
  for (int32_t j = 0; j < pr->numOfCols; ++j) {
M
Minglei Jin 已提交
324 325 326
    pRes[j] =
        taosMemoryCalloc(1, sizeof(SFirstLastRes) + pr->pSchema->columns[/*-1 == slotIds[j] ? 0 : */ slotIds[j]].bytes +
                                VARSTR_HEADER_SIZE);
327 328 329 330
    SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);
    p->ts = INT64_MIN;
  }

331
  pLastCols = taosArrayInit(pr->numOfCols, sizeof(SLastCol));
332 333 334 335 336
  if (pLastCols == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _end;
  }

337 338 339
  for (int32_t i = 0; i < pr->numOfCols; ++i) {
    int32_t          slotId = slotIds[i];
    struct STColumn* pCol = &pr->pSchema->columns[slotId];
340
    SLastCol         p = {.ts = INT64_MIN, .colVal.type = pCol->type, .colVal.flag = CV_FLAG_NULL};
341 342 343 344 345

    if (IS_VAR_DATA_TYPE(pCol->type)) {
      p.colVal.value.pData = taosMemoryCalloc(pCol->bytes, sizeof(char));
    }
    taosArrayPush(pLastCols, &p);
H
Haojun Liao 已提交
346
  }
347

348
  taosThreadMutexLock(&pr->readerMutex);
M
Minglei Jin 已提交
349
  code = tsdbTakeReadSnap2((STsdbReader*)pr, tsdbCacheQueryReseek, &pr->pReadSnap);
K
kailixu 已提交
350 351 352
  if (code != TSDB_CODE_SUCCESS) {
    goto _end;
  }
353

354
  int8_t ltype = (pr->type & CACHESCAN_RETRIEVE_LAST) >> 3;
355

356
  // retrieve the only one last row of all tables in the uid list.
H
Haojun Liao 已提交
357
  if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_SINGLE)) {
358
    int64_t st = taosGetTimestampUs();
X
Xiaoyu Wang 已提交
359
    int64_t totalLastTs = INT64_MAX;
360

361 362
    for (int32_t i = 0; i < pr->numOfTables; ++i) {
      STableKeyInfo* pKeyInfo = &pr->pTableList[i];
363

364 365
      tsdbCacheGetBatch(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
      // tsdbCacheGet(pr->pTsdb, pKeyInfo->uid, pRow, pr, ltype);
366
      if (TARRAY_SIZE(pRow) <= 0) {
M
Minglei Jin 已提交
367 368
        taosArrayClearEx(pRow, freeItem);
        // taosArrayClear(pRow);
369 370
        continue;
      }
371
      SLastCol* pColVal = taosArrayGet(pRow, 0);
M
Minglei Jin 已提交
372
      if (COL_VAL_IS_NONE(&pColVal->colVal)) {
M
Minglei Jin 已提交
373 374
        taosArrayClearEx(pRow, freeItem);
        // taosArrayClear(pRow);
375 376 377
        continue;
      }

H
Haojun Liao 已提交
378
      {
379
        bool    hasNotNullRow = true;
X
Xiaoyu Wang 已提交
380
        int64_t singleTableLastTs = INT64_MAX;
H
Haojun Liao 已提交
381
        for (int32_t k = 0; k < pr->numOfCols; ++k) {
382 383
          SLastCol* p = taosArrayGet(pLastCols, k);
          SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, k);
384

385 386 387 388
          if (pColVal->ts > p->ts) {
            if (!COL_VAL_IS_VALUE(&pColVal->colVal) && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
              if (!COL_VAL_IS_VALUE(&p->colVal)) {
                hasNotNullRow = false;
H
Haojun Liao 已提交
389
              }
390
              continue;
H
Haojun Liao 已提交
391
            }
392

393 394
            hasRes = true;
            p->ts = pColVal->ts;
395 396 397 398 399
            if (k == 0) {
              if (TARRAY_SIZE(pTableUidList) == 0) {
                taosArrayPush(pTableUidList, &pKeyInfo->uid);
              } else {
                taosArraySet(pTableUidList, 0, &pKeyInfo->uid);
400
              }
401
            }
H
Haojun Liao 已提交
402

403 404 405
            if (pColVal->ts < singleTableLastTs && HASTYPE(pr->type, CACHESCAN_RETRIEVE_LAST)) {
              singleTableLastTs = pColVal->ts;
            }
H
Haojun Liao 已提交
406

407 408 409 410 411
            if (!IS_VAR_DATA_TYPE(pColVal->colVal.type)) {
              p->colVal = pColVal->colVal;
            } else {
              if (COL_VAL_IS_VALUE(&pColVal->colVal)) {
                memcpy(p->colVal.value.pData, pColVal->colVal.value.pData, pColVal->colVal.value.nData);
H
Haojun Liao 已提交
412
              }
413 414 415 416 417

              p->colVal.value.nData = pColVal->colVal.value.nData;
              p->colVal.type = pColVal->colVal.type;
              p->colVal.flag = pColVal->colVal.flag;
              p->colVal.cid = pColVal->colVal.cid;
H
Haojun Liao 已提交
418 419
            }
          }
420
        }
421 422

        if (hasNotNullRow) {
X
Xiaoyu Wang 已提交
423
          if (INT64_MAX == totalLastTs || (INT64_MAX != singleTableLastTs && totalLastTs < singleTableLastTs)) {
X
Xiaoyu Wang 已提交
424
            totalLastTs = singleTableLastTs;
X
Xiaoyu Wang 已提交
425
          }
426 427
          double cost = (taosGetTimestampUs() - st) / 1000.0;
          if (cost > tsCacheLazyLoadThreshold) {
X
Xiaoyu Wang 已提交
428
            pr->lastTs = totalLastTs;
429
          }
430
        }
431
      }
432

M
Minglei Jin 已提交
433 434
      taosArrayClearEx(pRow, freeItem);
      // taosArrayClear(pRow);
435
    }
436 437

    if (hasRes) {
438
      saveOneRow(pLastCols, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
439
    }
H
Haojun Liao 已提交
440
  } else if (HASTYPE(pr->type, CACHESCAN_RETRIEVE_TYPE_ALL)) {
441
    for (int32_t i = pr->tableIndex; i < pr->numOfTables; ++i) {
442
      tb_uid_t uid = pr->pTableList[i].uid;
443

444
      tsdbCacheGetBatch(pr->pTsdb, uid, pRow, pr, ltype);
445
      if (TARRAY_SIZE(pRow) <= 0) {
M
Minglei Jin 已提交
446 447
        taosArrayClearEx(pRow, freeItem);
        // taosArrayClear(pRow);
448 449
        continue;
      }
M
Minglei Jin 已提交
450 451
      SLastCol* pColVal = (SLastCol*)taosArrayGet(pRow, 0);
      if (COL_VAL_IS_NONE(&pColVal->colVal)) {
M
Minglei Jin 已提交
452 453
        taosArrayClearEx(pRow, freeItem);
        // taosArrayClear(pRow);
454 455 456
        continue;
      }

457
      saveOneRow(pRow, pResBlock, pr, slotIds, dstSlotIds, pRes, pr->idstr);
M
Minglei Jin 已提交
458 459
      taosArrayClearEx(pRow, freeItem);
      // taosArrayClear(pRow);
460

461
      taosArrayPush(pTableUidList, &uid);
462

463
      ++pr->tableIndex;
464
      if (pResBlock->info.rows >= pResBlock->info.capacity) {
465
        goto _end;
466
      }
467 468
    }
  } else {
469
    code = TSDB_CODE_INVALID_PARA;
470 471
  }

472
_end:
473
  tsdbUntakeReadSnap2((STsdbReader*)pr, pr->pReadSnap, true);
474

M
Minglei Jin 已提交
475 476 477 478 479
  int64_t loadBlocks = 0;
  double  elapse = 0;
  pr->pLDataIterArray = destroySttBlockReader(pr->pLDataIterArray, &loadBlocks, &elapse);
  pr->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);

480
  taosThreadMutexUnlock(&pr->readerMutex);
481

H
Haojun Liao 已提交
482 483 484 485
  if (pRes != NULL) {
    for (int32_t j = 0; j < pr->numOfCols; ++j) {
      taosMemoryFree(pRes[j]);
    }
486 487 488
  }

  taosMemoryFree(pRes);
489 490
  // taosArrayDestroyEx(pRow, freeItem);
  taosArrayDestroy(pRow);
491
  taosArrayDestroyEx(pLastCols, freeItem);
492

493
  return code;
494
}