#4986

Committed 15 Mar 2026 08:32AM UTC coverage: 37.305% (-31.3%) from 68.601%

Build # #4986

Build Type

push

travis-ci

Committed by

tomchon

Commit Message

test: keep docs and unit test

Run Details

125478 of 336361 relevant lines covered (37.3%)

1134847.06 hits per line

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

3.79

/source/dnode/vnode/src/tsdb/tsdbScan.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 "tsdb.h"
#include "tsdbDataFileRW.h"
#include "tsdbFS2.h"
#include "tsdbSttFileRW.h"

typedef struct {
  STsdb    *tsdb;
  int32_t   fid;
  SVATaskID taskid;
  int64_t   scanSize;
} SScanArg;

typedef struct {
  int32_t   fid;
  SVATaskID taskId;
  int64_t   scanSize;
} SScanState;

struct SScanMonitor {
  int32_t          totalScanTasks;
  int64_t          startTimeSec;
  SArray          *tasks;
  int64_t          totalScanSize;
  int64_t          finishedScanSize;
  int64_t          lastUpdateFinishedSizeTime;
  volatile int32_t killed;
};

typedef struct {
  STsdb *tsdb;
  STFileSet *fset;
} SScanContext;

static void tsdbRemoveScanMonitorTask(STsdb *tsdb, SVATaskID *taskId);
static FORCE_INLINE bool tsdbScanTaskCanceled(STsdb *tsdb) { return atomic_load_32(&tsdb->pScanMonitor->killed) != 0; }

static int32_t tsdbScanContextInit(STsdb *tsdb, SScanContext **ppCtx) {
  int32_t       code = 0;
  int32_t       lino;
  SScanContext *ctx = (SScanContext *)taosMemoryCalloc(1, sizeof(SScanContext));
  if (ctx == NULL) {
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
  }

  ctx->tsdb = tsdb;
  *ppCtx = ctx;
_exit:
  if (code) {
    if (ctx) {
      taosMemoryFree(ctx);
    }
  }
  return code;
}

static void tsdbScanContextDestroy(SScanContext *ctx) {
  if (ctx == NULL) return;
  taosMemoryFree(ctx);
}

static void tsdbScanSttFile(STFileObj *fobj, SScanContext *pCtx) {
  struct SSttFileReader *reader = NULL;
  int32_t                code = TSDB_CODE_SUCCESS;
  char                   logPrefix[1024] = {0};
  SBlockData             blockData = {0};

  snprintf(logPrefix, sizeof(logPrefix), "[SCAN] fname:%s", fobj->fname);

  tsdbInfo("vgId:%d %s start scan stt file %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname);

  // Open the STT file reader
  SSttFileReaderConfig config = {
      .tsdb = pCtx->tsdb,
      .szPage = pCtx->tsdb->pVnode->config.tsdbPageSize,
      .file[0] = fobj->f[0],
  };
  code = tsdbSttFileReaderOpen(fobj->fname, &config, &reader);
  if (code) {
    tsdbError("vgId:%d %s failed to open stt file %s since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname,
              tstrerror(code));
    return;
  }

  // Do scan the stt file
  const TSttBlkArray *sttBlkArray = NULL;
  code = tsdbSttFileReadSttBlk(reader, &sttBlkArray);
  if (code) {
    tsdbError("vgId:%d %s failed to read stt blocks from %s since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix,
              fobj->fname, tstrerror(code));
    goto _exit;
  }

  // Loop to scan each stt block
  SSttBlk *sttBlk = NULL;
  TARRAY2_FOREACH_PTR(sttBlkArray, sttBlk) {
    code = tsdbSttFileReadBlockData(reader, sttBlk, &blockData);
    if (code) {
      tsdbError("vgId:%d %s failed to read block data from %s since %s, suid:%" PRId64 " min uid:%" PRId64
                " max uid:%" PRId64 " min ts:%" PRId64 " max ts:%" PRId64 " block offset:%" PRId64 " block size:%d",
                TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname, tstrerror(code), sttBlk->suid, sttBlk->minUid,
                sttBlk->maxUid, sttBlk->minKey, sttBlk->maxKey, sttBlk->bInfo.offset, sttBlk->bInfo.szBlock);
    } else {
      tsdbTrace("vgId:%d %s read block data from %s, suid:%" PRId64 " min uid:%" PRId64 " max uid:%" PRId64
                " min ts:%" PRId64 " max ts:%" PRId64 " block offset:%" PRId64 " block size:%d, nRow:%d",
                TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname, sttBlk->suid, sttBlk->minUid, sttBlk->maxUid,
                sttBlk->minKey, sttBlk->maxKey, sttBlk->bInfo.offset, sttBlk->bInfo.szBlock, blockData.nRow);
    }

    if (tsdbScanTaskCanceled(pCtx->tsdb)) {
      tsdbTrace("vgId:%d %s scan task is canceled, stop scanning stt file %s", TD_VID(pCtx->tsdb->pVnode), logPrefix,
                fobj->fname);
      goto _exit;
    }
  }

_exit:
  tBlockDataDestroy(&blockData);
  tsdbSttFileReaderClose(&reader);
  tsdbInfo("vgId:%d %s, finish scan stt file %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname);
}

static void tsdbScanDataFile(SScanContext *pCtx) {
  SDataFileReader *reader = NULL;
  int32_t          code = TSDB_CODE_SUCCESS;
  SBlockData       blockData = {0};
  STFileObj       *fobj = NULL;
  char             logPrefix[1024] = {0};

  if (pCtx->fset->farr[TSDB_FTYPE_DATA] == NULL) {
    return;
  }

  snprintf(logPrefix, sizeof(logPrefix), "[SCAN] data file:%s", pCtx->fset->farr[TSDB_FTYPE_DATA]->fname);

  tsdbInfo("vgId:%d %s start scan data file", TD_VID(pCtx->tsdb->pVnode), logPrefix);

  // Open the data file reader
  SDataFileReaderConfig config = {
      .tsdb = pCtx->tsdb,
      .szPage = pCtx->tsdb->pVnode->config.tsdbPageSize,
  };
  for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = pCtx->fset->farr[ftype], 1); ftype++) {
    if (fobj == NULL) {
      continue;
    }

    config.files[ftype].exist = true;
    config.files[ftype].file = fobj->f[0];
  }

  // Open the data file reader
  code = tsdbDataFileReaderOpen(NULL, &config, &reader);
  if (code) {
    tsdbError("vgId:%d %s failed to open data file since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code));
    goto _exit;
  }

  // Load BRIN index
  const TBrinBlkArray *brinBlkArray = NULL;
  code = tsdbDataFileReadBrinBlk(reader, &brinBlkArray);
  if (code) {
    tsdbError("vgId:%d %s failed to read BRIN blocks since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code));
    goto _exit;
  }

  // Loop each BRIN block
  for (int32_t i = 0; i < TARRAY2_SIZE(brinBlkArray); ++i) {
    const SBrinBlk *brinBlk = TARRAY2_GET_PTR(brinBlkArray, i);

    // Load BRIN block
    SBrinBlock brinBlock = {0};
    code = tsdbDataFileReadBrinBlock(reader, brinBlk, &brinBlock);
    if (code) {
      tsdbError("vgId:%d %s failed to read BRIN block since %s, min table id:(%" PRId64 ", %" PRId64
                ") max table id:(%" PRId64 ", %" PRId64 ") BRIN block offset:%" PRId64 " BRIN block size:%" PRId64,
                TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code), brinBlk->minTbid.suid, brinBlk->minTbid.uid,
                brinBlk->maxTbid.suid, brinBlk->maxTbid.uid, brinBlk->dp[0].offset, brinBlk->dp[0].size);
      continue;
    }

    // Loop each BRIN record
    SBrinRecord record = {0};
    for (int32_t j = 0; j < BRIN_BLOCK_SIZE(&brinBlock); ++j) {
      code = tBrinBlockGet(&brinBlock, j, &record);

      // Load block data
      code = tsdbDataFileReadBlockData(reader, &record, &blockData);
      if (code) {
        tsdbError("vgId:%d %s failed to read block data since %s, suid:%" PRId64 " uid:%" PRId64 " min ts:%" PRId64
                  " max ts:%" PRId64 " min version:%" PRId64 " max version:%" PRId64 " block offset:%" PRId64
                  " block size:%d block key size:%d number rows:%d count:%d",
                  TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code), record.suid, record.uid,
                  record.firstKey.key.ts, record.lastKey.key.ts, record.minVer, record.maxVer, record.blockOffset,
                  record.blockSize, record.blockKeySize, record.numRow, record.count);
        continue;
      }
    }
  }

_exit:
  tBlockDataDestroy(&blockData);
  tsdbDataFileReaderClose(&reader);
  tsdbInfo("vgId:%d %s finish scan data file", TD_VID(pCtx->tsdb->pVnode), logPrefix);
}

static void tsdbDoScanFileSet(SScanContext *pCtx) {
  // Do scan the STT files
  SSttLvl   *lvl = NULL;
  STFileObj *fobj = NULL;
  TARRAY2_FOREACH(pCtx->fset->lvlArr, lvl) {
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      if (tsdbScanTaskCanceled(pCtx->tsdb)) {
        return;
      }
      tsdbScanSttFile(fobj, pCtx);
    }
  }

  if (tsdbScanTaskCanceled(pCtx->tsdb)) {
    return;
  }

  // Do scan the data file
  tsdbScanDataFile(pCtx);
}

static void scanArgDestroy(SScanArg *arg);

static int32_t tsdbScanFileSet(void *arg) {
  int32_t   code = TSDB_CODE_SUCCESS;
  SScanArg *scanArg = (SScanArg *)arg;
  STsdb    *tsdb = scanArg->tsdb;
  SScanContext ctx = {
      .tsdb = tsdb,
  };

  tsdbInfo("vgId:%d, start scan file set, fid:%d, taskId:%" PRId64, TD_VID(tsdb->pVnode), scanArg->fid,
           scanArg->taskid.id);

  // Check if task is canceled
  if (tsdbScanTaskCanceled(tsdb)) {
    code = TSDB_CODE_SUCCESS;
    goto _exit;
  }

  // Get the file set
  (void)taosThreadMutexLock(&tsdb->mutex);

  if (tsdb->bgTaskDisabled) {
    tsdbInfo("vgId:%d, background task is disabled, skip scan", TD_VID(tsdb->pVnode));
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    code = TSDB_CODE_SUCCESS;
    goto _exit;
  }

  STFileSet *fset = NULL;
  tsdbFSGetFSet(tsdb->pFS, scanArg->fid, &fset);
  if (fset == NULL) {
    tsdbInfo("vgId:%d, fid:%d not found, skip scan", TD_VID(tsdb->pVnode), scanArg->fid);
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    code = TSDB_CODE_SUCCESS;
    goto _exit;
  }

  code = tsdbTFileSetInitRef(tsdb, fset, &ctx.fset);
  if (code) {
    tsdbError("vgId:%d, fid:%d, tsdbTFileSetInitRef failed, code:%d", TD_VID(tsdb->pVnode), scanArg->fid, code);
    (void)taosThreadMutexUnlock(&tsdb->mutex);
    goto _exit;
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);

  // Do scan the file set
  tsdbDoScanFileSet(&ctx);

_exit:
  tsdbInfo("vgId:%d, finish scan file set, fid:%d, taskId:%" PRId64 ", code:%d", TD_VID(tsdb->pVnode), scanArg->fid,
           scanArg->taskid.id, code);
  tsdbTFileSetClear(&ctx.fset);
  tsdbRemoveScanMonitorTask(tsdb, &scanArg->taskid);
  scanArgDestroy(scanArg);
  return code;
}

static void scanArgDestroy(SScanArg *arg) {
  if (arg == NULL) return;
  taosMemoryFree(arg);
}

static void tsdbScanFileSetCancel(void *arg) { scanArgDestroy((SScanArg *)arg); }

static int64_t tsdbGetFileSetScanSize(STFileSet *fset) {
  int64_t    size = 0;
  STFileObj *fobj;
  for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = fset->farr[ftype], 1); ftype++) {
    if (fobj == NULL) continue;
    size += fobj->f[0].size;
  }

  SSttLvl *lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    TARRAY2_FOREACH(lvl->fobjArr, fobj) { size += fobj->f[0].size; }
  }
  return size;
}

static void tsdbAddScanMonitorTask(STsdb *tsdb, SScanArg *arg) {
  if (tsdb->pScanMonitor == NULL) {
    return;
  }

  // Init a new task
  if (taosArrayGetSize(tsdb->pScanMonitor->tasks) == 0) {
    tsdb->pScanMonitor->startTimeSec = taosGetTimestampSec();
    tsdb->pScanMonitor->totalScanTasks = 0;
    tsdb->pScanMonitor->totalScanSize = 0;
    tsdb->pScanMonitor->finishedScanSize = 0;
    tsdb->pScanMonitor->lastUpdateFinishedSizeTime = tsdb->pScanMonitor->startTimeSec;
    tsdb->pScanMonitor->killed = 0;
  }

  if (NULL == tsdb->pScanMonitor->tasks) {
    tsdb->pScanMonitor->tasks = taosArrayInit(64, sizeof(SScanState));
    if (tsdb->pScanMonitor->tasks == NULL) {
      tsdbError("vid:%d, failed to allocate memory for scan monitor tasks", TD_VID(tsdb->pVnode));
      return;
    }
  }

  SScanState state = {
      .fid = arg->fid,
      .taskId = arg->taskid,
      .scanSize = arg->scanSize,
  };

  if (NULL == taosArrayPush(tsdb->pScanMonitor->tasks, (const void *)&state)) {
    tsdbError("vid:%d, failed to add scan monitor task for fid:%d", TD_VID(tsdb->pVnode), arg->fid);
    return;
  }

  tsdb->pScanMonitor->totalScanTasks++;
  tsdb->pScanMonitor->totalScanSize += state.scanSize;
  tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is added to scan monitor", TD_VID(tsdb->pVnode),
            state.fid, state.taskId.id);
  return;
}

static void tsdbRemoveScanMonitorTask(STsdb *tsdb, SVATaskID *taskId) {
  (void)taosThreadMutexLock(&tsdb->mutex);

  for (int32_t i = 0; i < taosArrayGetSize(tsdb->pScanMonitor->tasks); i++) {
    SScanState *state = (SScanState *)taosArrayGet(tsdb->pScanMonitor->tasks, i);
    if (state->taskId.async == taskId->async && state->taskId.id == taskId->id) {
      tsdbInfo("vid:%d, fid:%d, taskId:%" PRId64 " is removed from scan monitor", TD_VID(tsdb->pVnode), state->fid,
               taskId->id);
      tsdb->pScanMonitor->finishedScanSize += state->scanSize;
      tsdb->pScanMonitor->lastUpdateFinishedSizeTime = taosGetTimestampSec();
      taosArrayRemove(tsdb->pScanMonitor->tasks, i);
      break;
    }
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);
}

static int32_t tsdbAsyncScanImpl(STsdb *tsdb, const STimeWindow *tw) {
  int32_t code = 0;
  int32_t lino = 0;

  if (tsdb->bgTaskDisabled) {
    tsdbInfo("vgId:%d, background task is disabled, skip scan", TD_VID(tsdb->pVnode));
    return 0;
  }

  int32_t minFid = tsdbKeyFid(tw->skey, tsdb->keepCfg.days, tsdb->keepCfg.precision);
  int32_t maxFid = tsdbKeyFid(tw->ekey, tsdb->keepCfg.days, tsdb->keepCfg.precision);

  STFileSet *fset;
  TARRAY2_FOREACH(tsdb->pFS->fSetArr, fset) {
    if (fset->fid < minFid || fset->fid > maxFid) {
      continue;
    }

    SScanArg *arg = taosMemoryCalloc(1, sizeof(*arg));
    if (arg == NULL) {
      code = terrno;
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    arg->tsdb = tsdb;
    arg->fid = fset->fid;
    arg->scanSize = tsdbGetFileSetScanSize(fset);
    code = vnodeAsync(SCAN_TASK_ASYNC, EVA_PRIORITY_NORMAL, tsdbScanFileSet, tsdbScanFileSetCancel, arg, &arg->taskid);
    if (code) {
      scanArgDestroy(arg);
      TSDB_CHECK_CODE(code, lino, _exit);
    }

    tsdbAddScanMonitorTask(tsdb, arg);
  }

_exit:
  if (code) {
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(tsdb->pVnode), __func__, lino, tstrerror(code));
  }
  return code;
}

int32_t tsdbAsyncScan(STsdb *tsdb, const STimeWindow *tw) {
  int32_t code = 0;
  (void)taosThreadMutexLock(&tsdb->mutex);
  code = tsdbAsyncScanImpl(tsdb, tw);
  (void)taosThreadMutexUnlock(&tsdb->mutex);
  return code;
}

void tsdbCancelScanTask(STsdb *tsdb) {
  if (tsdb->pScanMonitor == NULL) {
    return;
  }

  (void)taosThreadMutexLock(&tsdb->mutex);

  atomic_store_32(&tsdb->pScanMonitor->killed, 1);
  int32_t i = 0;
  while (true) {
    int32_t arraySize = taosArrayGetSize(tsdb->pScanMonitor->tasks);

    if (i >= arraySize) {
      break;
    }

    if (i < arraySize) {
      SScanState *state = (SScanState *)taosArrayGet(tsdb->pScanMonitor->tasks, i);
      if (vnodeACancel(&state->taskId) == 0) {
        tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is cancelled and removed from scan monitor", TD_VID(tsdb->pVnode),
                  state->fid, state->taskId.id);
        taosArrayRemove(tsdb->pScanMonitor->tasks, i);
      } else {
        i++;
      }
    }
  }

  (void)taosThreadMutexUnlock(&tsdb->mutex);
}

void tsdbScanMonitorGetInfo(STsdb *tsdb, SQueryScanProgressRsp *rsp) {
  (void)taosThreadMutexLock(&tsdb->mutex);
  rsp->scanId = 0;
  rsp->vgId = TD_VID(tsdb->pVnode);
  rsp->numberFileset = tsdb->pScanMonitor->totalScanTasks;
  rsp->finished = rsp->numberFileset - taosArrayGetSize(tsdb->pScanMonitor->tasks);
  // calculate progress
  if (tsdb->pScanMonitor->totalScanSize > 0) {
    rsp->progress = tsdb->pScanMonitor->finishedScanSize * 100 / tsdb->pScanMonitor->totalScanSize;
  } else {
    rsp->progress = 0;
  }
  // calculate estimated remaining time
  int64_t elapsed = tsdb->pScanMonitor->lastUpdateFinishedSizeTime - tsdb->pScanMonitor->startTimeSec;
  if (rsp->progress > 0 && elapsed > 0) {
    rsp->remainingTime = elapsed * (100 - rsp->progress) / rsp->progress;
  } else {
    rsp->remainingTime = tsdb->pScanMonitor->totalScanSize / (20 * 1024 * 1024);  // suppose 20MB/s
  }
  (void)taosThreadMutexUnlock(&tsdb->mutex);
}

int32_t tsdbScanMonitorOpen(STsdb *tsdb) {
  tsdb->pScanMonitor = (struct SScanMonitor *)taosMemoryCalloc(1, sizeof(struct SScanMonitor));
  if (NULL == tsdb->pScanMonitor) {
    return terrno;
  }
  return TSDB_CODE_SUCCESS;
}

void tsdbScanMonitorClose(STsdb *tsdb) {
  if (tsdb->pScanMonitor) {
    taosArrayDestroy(tsdb->pScanMonitor->tasks);
    taosMemoryFree(tsdb->pScanMonitor);
    tsdb->pScanMonitor = NULL;
  }
}

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 "tsdb.h"
17	#include "tsdbDataFileRW.h"
18	#include "tsdbFS2.h"
19	#include "tsdbSttFileRW.h"
20
21	typedef struct {
22	STsdb *tsdb;
23	int32_t fid;
24	SVATaskID taskid;
25	int64_t scanSize;
26	} SScanArg;
27
28	typedef struct {
29	int32_t fid;
30	SVATaskID taskId;
31	int64_t scanSize;
32	} SScanState;
33
34	struct SScanMonitor {
35	int32_t totalScanTasks;
36	int64_t startTimeSec;
37	SArray *tasks;
38	int64_t totalScanSize;
39	int64_t finishedScanSize;
40	int64_t lastUpdateFinishedSizeTime;
41	volatile int32_t killed;
42	};
43
44	typedef struct {
45	STsdb *tsdb;
46	STFileSet *fset;
47	} SScanContext;
48
49	static void tsdbRemoveScanMonitorTask(STsdb tsdb, SVATaskID taskId);
50	static FORCE_INLINE bool tsdbScanTaskCanceled(STsdb *tsdb) { return atomic_load_32(&tsdb->pScanMonitor->killed) != 0; }	×
51
52	static int32_t tsdbScanContextInit(STsdb tsdb, SScanContext *ppCtx) {	×
53	int32_t code = 0;	×
54	int32_t lino;
55	SScanContext ctx = (SScanContext )taosMemoryCalloc(1, sizeof(SScanContext));	×
56	if (ctx == NULL) {	×
57	TSDB_CHECK_CODE(code = terrno, lino, _exit);	×
58	}
59
60	ctx->tsdb = tsdb;	×
61	*ppCtx = ctx;	×
62	_exit:	×
63	if (code) {	×
64	if (ctx) {	×
65	taosMemoryFree(ctx);	×
66	}
67	}
68	return code;	×
69	}
70
71	static void tsdbScanContextDestroy(SScanContext *ctx) {	×
72	if (ctx == NULL) return;	×
73	taosMemoryFree(ctx);	×
74	}
75
76	static void tsdbScanSttFile(STFileObj fobj, SScanContext pCtx) {	×
77	struct SSttFileReader *reader = NULL;	×
78	int32_t code = TSDB_CODE_SUCCESS;	×
79	char logPrefix[1024] = {0};	×
80	SBlockData blockData = {0};	×
81
82	snprintf(logPrefix, sizeof(logPrefix), "[SCAN] fname:%s", fobj->fname);	×
83
84	tsdbInfo("vgId:%d %s start scan stt file %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname);	×
85
86	// Open the STT file reader
87	SSttFileReaderConfig config = {	×
88	.tsdb = pCtx->tsdb,	×
89	.szPage = pCtx->tsdb->pVnode->config.tsdbPageSize,	×
90	.file[0] = fobj->f[0],
91	};
92	code = tsdbSttFileReaderOpen(fobj->fname, &config, &reader);	×
93	if (code) {	×
94	tsdbError("vgId:%d %s failed to open stt file %s since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname,	×
95	tstrerror(code));
96	return;	×
97	}
98
99	// Do scan the stt file
100	const TSttBlkArray *sttBlkArray = NULL;	×
101	code = tsdbSttFileReadSttBlk(reader, &sttBlkArray);	×
102	if (code) {	×
103	tsdbError("vgId:%d %s failed to read stt blocks from %s since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix,	×
104	fobj->fname, tstrerror(code));
105	goto _exit;	×
106	}
107
108	// Loop to scan each stt block
109	SSttBlk *sttBlk = NULL;	×
110	TARRAY2_FOREACH_PTR(sttBlkArray, sttBlk) {	×
111	code = tsdbSttFileReadBlockData(reader, sttBlk, &blockData);	×
112	if (code) {	×
113	tsdbError("vgId:%d %s failed to read block data from %s since %s, suid:%" PRId64 " min uid:%" PRId64	×
114	" max uid:%" PRId64 " min ts:%" PRId64 " max ts:%" PRId64 " block offset:%" PRId64 " block size:%d",
115	TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname, tstrerror(code), sttBlk->suid, sttBlk->minUid,
116	sttBlk->maxUid, sttBlk->minKey, sttBlk->maxKey, sttBlk->bInfo.offset, sttBlk->bInfo.szBlock);
117	} else {
118	tsdbTrace("vgId:%d %s read block data from %s, suid:%" PRId64 " min uid:%" PRId64 " max uid:%" PRId64	×
119	" min ts:%" PRId64 " max ts:%" PRId64 " block offset:%" PRId64 " block size:%d, nRow:%d",
120	TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname, sttBlk->suid, sttBlk->minUid, sttBlk->maxUid,
121	sttBlk->minKey, sttBlk->maxKey, sttBlk->bInfo.offset, sttBlk->bInfo.szBlock, blockData.nRow);
122	}
123
124	if (tsdbScanTaskCanceled(pCtx->tsdb)) {	×
125	tsdbTrace("vgId:%d %s scan task is canceled, stop scanning stt file %s", TD_VID(pCtx->tsdb->pVnode), logPrefix,	×
126	fobj->fname);
127	goto _exit;	×
128	}
129	}
130
131	_exit:	×
132	tBlockDataDestroy(&blockData);	×
133	tsdbSttFileReaderClose(&reader);	×
134	tsdbInfo("vgId:%d %s, finish scan stt file %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, fobj->fname);	×
135	}
136
137	static void tsdbScanDataFile(SScanContext *pCtx) {	×
138	SDataFileReader *reader = NULL;	×
139	int32_t code = TSDB_CODE_SUCCESS;	×
140	SBlockData blockData = {0};	×
141	STFileObj *fobj = NULL;	×
142	char logPrefix[1024] = {0};	×
143
144	if (pCtx->fset->farr[TSDB_FTYPE_DATA] == NULL) {	×
145	return;	×
146	}
147
148	snprintf(logPrefix, sizeof(logPrefix), "[SCAN] data file:%s", pCtx->fset->farr[TSDB_FTYPE_DATA]->fname);	×
149
150	tsdbInfo("vgId:%d %s start scan data file", TD_VID(pCtx->tsdb->pVnode), logPrefix);	×
151
152	// Open the data file reader
153	SDataFileReaderConfig config = {	×
154	.tsdb = pCtx->tsdb,	×
155	.szPage = pCtx->tsdb->pVnode->config.tsdbPageSize,	×
156	};
157	for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = pCtx->fset->farr[ftype], 1); ftype++) {	×
158	if (fobj == NULL) {	×
159	continue;	×
160	}
161
162	config.files[ftype].exist = true;	×
163	config.files[ftype].file = fobj->f[0];	×
164	}
165
166	// Open the data file reader
167	code = tsdbDataFileReaderOpen(NULL, &config, &reader);	×
168	if (code) {	×
169	tsdbError("vgId:%d %s failed to open data file since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code));	×
170	goto _exit;	×
171	}
172
173	// Load BRIN index
174	const TBrinBlkArray *brinBlkArray = NULL;	×
175	code = tsdbDataFileReadBrinBlk(reader, &brinBlkArray);	×
176	if (code) {	×
177	tsdbError("vgId:%d %s failed to read BRIN blocks since %s", TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code));	×
178	goto _exit;	×
179	}
180
181	// Loop each BRIN block
182	for (int32_t i = 0; i < TARRAY2_SIZE(brinBlkArray); ++i) {	×
183	const SBrinBlk *brinBlk = TARRAY2_GET_PTR(brinBlkArray, i);	×
184
185	// Load BRIN block
186	SBrinBlock brinBlock = {0};	×
187	code = tsdbDataFileReadBrinBlock(reader, brinBlk, &brinBlock);	×
188	if (code) {	×
189	tsdbError("vgId:%d %s failed to read BRIN block since %s, min table id:(%" PRId64 ", %" PRId64	×
190	") max table id:(%" PRId64 ", %" PRId64 ") BRIN block offset:%" PRId64 " BRIN block size:%" PRId64,
191	TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code), brinBlk->minTbid.suid, brinBlk->minTbid.uid,
192	brinBlk->maxTbid.suid, brinBlk->maxTbid.uid, brinBlk->dp[0].offset, brinBlk->dp[0].size);
193	continue;	×
194	}
195
196	// Loop each BRIN record
197	SBrinRecord record = {0};	×
198	for (int32_t j = 0; j < BRIN_BLOCK_SIZE(&brinBlock); ++j) {	×
199	code = tBrinBlockGet(&brinBlock, j, &record);	×
200
201	// Load block data
202	code = tsdbDataFileReadBlockData(reader, &record, &blockData);	×
203	if (code) {	×
204	tsdbError("vgId:%d %s failed to read block data since %s, suid:%" PRId64 " uid:%" PRId64 " min ts:%" PRId64	×
205	" max ts:%" PRId64 " min version:%" PRId64 " max version:%" PRId64 " block offset:%" PRId64
206	" block size:%d block key size:%d number rows:%d count:%d",
207	TD_VID(pCtx->tsdb->pVnode), logPrefix, tstrerror(code), record.suid, record.uid,
208	record.firstKey.key.ts, record.lastKey.key.ts, record.minVer, record.maxVer, record.blockOffset,
209	record.blockSize, record.blockKeySize, record.numRow, record.count);
210	continue;	×
211	}
212	}
213	}
214
215	_exit:	×
216	tBlockDataDestroy(&blockData);	×
217	tsdbDataFileReaderClose(&reader);	×
218	tsdbInfo("vgId:%d %s finish scan data file", TD_VID(pCtx->tsdb->pVnode), logPrefix);	×
219	}
220
221	static void tsdbDoScanFileSet(SScanContext *pCtx) {	×
222	// Do scan the STT files
223	SSttLvl *lvl = NULL;	×
224	STFileObj *fobj = NULL;	×
225	TARRAY2_FOREACH(pCtx->fset->lvlArr, lvl) {	×
226	TARRAY2_FOREACH(lvl->fobjArr, fobj) {	×
227	if (tsdbScanTaskCanceled(pCtx->tsdb)) {	×
228	return;	×
229	}
230	tsdbScanSttFile(fobj, pCtx);	×
231	}
232	}
233
234	if (tsdbScanTaskCanceled(pCtx->tsdb)) {	×
235	return;	×
236	}
237
238	// Do scan the data file
239	tsdbScanDataFile(pCtx);	×
240	}
241
242	static void scanArgDestroy(SScanArg *arg);
243
244	static int32_t tsdbScanFileSet(void *arg) {	×
245	int32_t code = TSDB_CODE_SUCCESS;	×
246	SScanArg scanArg = (SScanArg )arg;	×
247	STsdb *tsdb = scanArg->tsdb;	×
248	SScanContext ctx = {	×
249	.tsdb = tsdb,
250	};
251
252	tsdbInfo("vgId:%d, start scan file set, fid:%d, taskId:%" PRId64, TD_VID(tsdb->pVnode), scanArg->fid,	×
253	scanArg->taskid.id);
254
255	// Check if task is canceled
256	if (tsdbScanTaskCanceled(tsdb)) {	×
257	code = TSDB_CODE_SUCCESS;	×
258	goto _exit;	×
259	}
260
261	// Get the file set
262	(void)taosThreadMutexLock(&tsdb->mutex);	×
263
264	if (tsdb->bgTaskDisabled) {	×
265	tsdbInfo("vgId:%d, background task is disabled, skip scan", TD_VID(tsdb->pVnode));	×
266	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
267	code = TSDB_CODE_SUCCESS;	×
268	goto _exit;	×
269	}
270
271	STFileSet *fset = NULL;	×
272	tsdbFSGetFSet(tsdb->pFS, scanArg->fid, &fset);	×
273	if (fset == NULL) {	×
274	tsdbInfo("vgId:%d, fid:%d not found, skip scan", TD_VID(tsdb->pVnode), scanArg->fid);	×
275	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
276	code = TSDB_CODE_SUCCESS;	×
277	goto _exit;	×
278	}
279
280	code = tsdbTFileSetInitRef(tsdb, fset, &ctx.fset);	×
281	if (code) {	×
282	tsdbError("vgId:%d, fid:%d, tsdbTFileSetInitRef failed, code:%d", TD_VID(tsdb->pVnode), scanArg->fid, code);	×
283	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
284	goto _exit;	×
285	}
286
287	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
288
289	// Do scan the file set
290	tsdbDoScanFileSet(&ctx);	×
291
292	_exit:	×
293	tsdbInfo("vgId:%d, finish scan file set, fid:%d, taskId:%" PRId64 ", code:%d", TD_VID(tsdb->pVnode), scanArg->fid,	×
294	scanArg->taskid.id, code);
295	tsdbTFileSetClear(&ctx.fset);	×
296	tsdbRemoveScanMonitorTask(tsdb, &scanArg->taskid);	×
297	scanArgDestroy(scanArg);	×
298	return code;	×
299	}
300
301	static void scanArgDestroy(SScanArg *arg) {	×
302	if (arg == NULL) return;	×
303	taosMemoryFree(arg);	×
304	}
305
306	static void tsdbScanFileSetCancel(void arg) { scanArgDestroy((SScanArg )arg); }	×
307
308	static int64_t tsdbGetFileSetScanSize(STFileSet *fset) {	×
309	int64_t size = 0;	×
310	STFileObj *fobj;
311	for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX && (fobj = fset->farr[ftype], 1); ftype++) {	×
312	if (fobj == NULL) continue;	×
313	size += fobj->f[0].size;	×
314	}
315
316	SSttLvl *lvl;
317	TARRAY2_FOREACH(fset->lvlArr, lvl) {	×
318	TARRAY2_FOREACH(lvl->fobjArr, fobj) { size += fobj->f[0].size; }	×
319	}
320	return size;	×
321	}
322
323	static void tsdbAddScanMonitorTask(STsdb tsdb, SScanArg arg) {	×
324	if (tsdb->pScanMonitor == NULL) {	×
325	return;	×
326	}
327
328	// Init a new task
329	if (taosArrayGetSize(tsdb->pScanMonitor->tasks) == 0) {	×
330	tsdb->pScanMonitor->startTimeSec = taosGetTimestampSec();	×
331	tsdb->pScanMonitor->totalScanTasks = 0;	×
332	tsdb->pScanMonitor->totalScanSize = 0;	×
333	tsdb->pScanMonitor->finishedScanSize = 0;	×
334	tsdb->pScanMonitor->lastUpdateFinishedSizeTime = tsdb->pScanMonitor->startTimeSec;	×
335	tsdb->pScanMonitor->killed = 0;	×
336	}
337
338	if (NULL == tsdb->pScanMonitor->tasks) {	×
339	tsdb->pScanMonitor->tasks = taosArrayInit(64, sizeof(SScanState));	×
340	if (tsdb->pScanMonitor->tasks == NULL) {	×
341	tsdbError("vid:%d, failed to allocate memory for scan monitor tasks", TD_VID(tsdb->pVnode));	×
342	return;	×
343	}
344	}
345
346	SScanState state = {	×
347	.fid = arg->fid,	×
348	.taskId = arg->taskid,
349	.scanSize = arg->scanSize,	×
350	};
351
352	if (NULL == taosArrayPush(tsdb->pScanMonitor->tasks, (const void *)&state)) {	×
353	tsdbError("vid:%d, failed to add scan monitor task for fid:%d", TD_VID(tsdb->pVnode), arg->fid);	×
354	return;	×
355	}
356
357	tsdb->pScanMonitor->totalScanTasks++;	×
358	tsdb->pScanMonitor->totalScanSize += state.scanSize;	×
359	tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is added to scan monitor", TD_VID(tsdb->pVnode),	×
360	state.fid, state.taskId.id);
361	return;	×
362	}
363
364	static void tsdbRemoveScanMonitorTask(STsdb tsdb, SVATaskID taskId) {	×
365	(void)taosThreadMutexLock(&tsdb->mutex);	×
366
367	for (int32_t i = 0; i < taosArrayGetSize(tsdb->pScanMonitor->tasks); i++) {	×
368	SScanState state = (SScanState )taosArrayGet(tsdb->pScanMonitor->tasks, i);	×
369	if (state->taskId.async == taskId->async && state->taskId.id == taskId->id) {	×
370	tsdbInfo("vid:%d, fid:%d, taskId:%" PRId64 " is removed from scan monitor", TD_VID(tsdb->pVnode), state->fid,	×
371	taskId->id);
372	tsdb->pScanMonitor->finishedScanSize += state->scanSize;	×
373	tsdb->pScanMonitor->lastUpdateFinishedSizeTime = taosGetTimestampSec();	×
374	taosArrayRemove(tsdb->pScanMonitor->tasks, i);	×
375	break;	×
376	}
377	}
378
379	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
380	}	×
381
382	static int32_t tsdbAsyncScanImpl(STsdb tsdb, const STimeWindow tw) {	×
383	int32_t code = 0;	×
384	int32_t lino = 0;	×
385
386	if (tsdb->bgTaskDisabled) {	×
387	tsdbInfo("vgId:%d, background task is disabled, skip scan", TD_VID(tsdb->pVnode));	×
388	return 0;	×
389	}
390
391	int32_t minFid = tsdbKeyFid(tw->skey, tsdb->keepCfg.days, tsdb->keepCfg.precision);	×
392	int32_t maxFid = tsdbKeyFid(tw->ekey, tsdb->keepCfg.days, tsdb->keepCfg.precision);	×
393
394	STFileSet *fset;
395	TARRAY2_FOREACH(tsdb->pFS->fSetArr, fset) {	×
396	if (fset->fid < minFid \|\| fset->fid > maxFid) {	×
397	continue;	×
398	}
399
400	SScanArg arg = taosMemoryCalloc(1, sizeof(arg));	×
401	if (arg == NULL) {	×
402	code = terrno;	×
403	TSDB_CHECK_CODE(code, lino, _exit);	×
404	}
405
406	arg->tsdb = tsdb;	×
407	arg->fid = fset->fid;	×
408	arg->scanSize = tsdbGetFileSetScanSize(fset);	×
409	code = vnodeAsync(SCAN_TASK_ASYNC, EVA_PRIORITY_NORMAL, tsdbScanFileSet, tsdbScanFileSetCancel, arg, &arg->taskid);	×
410	if (code) {	×
411	scanArgDestroy(arg);	×
412	TSDB_CHECK_CODE(code, lino, _exit);	×
413	}
414
415	tsdbAddScanMonitorTask(tsdb, arg);	×
416	}
417
418	_exit:	×
419	if (code) {	×
420	tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(tsdb->pVnode), __func__, lino, tstrerror(code));	×
421	}
422	return code;	×
423	}
424
425	int32_t tsdbAsyncScan(STsdb tsdb, const STimeWindow tw) {	×
426	int32_t code = 0;	×
427	(void)taosThreadMutexLock(&tsdb->mutex);	×
428	code = tsdbAsyncScanImpl(tsdb, tw);	×
429	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
430	return code;	×
431	}
432
433	void tsdbCancelScanTask(STsdb *tsdb) {	×
434	if (tsdb->pScanMonitor == NULL) {	×
435	return;	×
436	}
437
438	(void)taosThreadMutexLock(&tsdb->mutex);	×
439
440	atomic_store_32(&tsdb->pScanMonitor->killed, 1);	×
441	int32_t i = 0;	×
442	while (true) {	×
443	int32_t arraySize = taosArrayGetSize(tsdb->pScanMonitor->tasks);	×
444
445	if (i >= arraySize) {	×
446	break;	×
447	}
448
449	if (i < arraySize) {	×
450	SScanState state = (SScanState )taosArrayGet(tsdb->pScanMonitor->tasks, i);	×
451	if (vnodeACancel(&state->taskId) == 0) {	×
452	tsdbDebug("vid:%d, fid:%d, taskId:%" PRId64 " is cancelled and removed from scan monitor", TD_VID(tsdb->pVnode),	×
453	state->fid, state->taskId.id);
454	taosArrayRemove(tsdb->pScanMonitor->tasks, i);	×
455	} else {
456	i++;	×
457	}
458	}
459	}
460
461	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
462	}
463
464	void tsdbScanMonitorGetInfo(STsdb tsdb, SQueryScanProgressRsp rsp) {	×
465	(void)taosThreadMutexLock(&tsdb->mutex);	×
466	rsp->scanId = 0;	×
467	rsp->vgId = TD_VID(tsdb->pVnode);	×
468	rsp->numberFileset = tsdb->pScanMonitor->totalScanTasks;	×
469	rsp->finished = rsp->numberFileset - taosArrayGetSize(tsdb->pScanMonitor->tasks);	×
470	// calculate progress
471	if (tsdb->pScanMonitor->totalScanSize > 0) {	×
472	rsp->progress = tsdb->pScanMonitor->finishedScanSize * 100 / tsdb->pScanMonitor->totalScanSize;	×
473	} else {
474	rsp->progress = 0;	×
475	}
476	// calculate estimated remaining time
477	int64_t elapsed = tsdb->pScanMonitor->lastUpdateFinishedSizeTime - tsdb->pScanMonitor->startTimeSec;	×
478	if (rsp->progress > 0 && elapsed > 0) {	×
479	rsp->remainingTime = elapsed * (100 - rsp->progress) / rsp->progress;	×
480	} else {
481	rsp->remainingTime = tsdb->pScanMonitor->totalScanSize / (20 * 1024 * 1024); // suppose 20MB/s	×
482	}
483	(void)taosThreadMutexUnlock(&tsdb->mutex);	×
484	}	×
485
486	int32_t tsdbScanMonitorOpen(STsdb *tsdb) {	40✔
487	tsdb->pScanMonitor = (struct SScanMonitor *)taosMemoryCalloc(1, sizeof(struct SScanMonitor));	40✔
488	if (NULL == tsdb->pScanMonitor) {	40✔
489	return terrno;	×
490	}
491	return TSDB_CODE_SUCCESS;	40✔
492	}
493
494	void tsdbScanMonitorClose(STsdb *tsdb) {	40✔
495	if (tsdb->pScanMonitor) {	40✔
496	taosArrayDestroy(tsdb->pScanMonitor->tasks);	40✔
497	taosMemoryFree(tsdb->pScanMonitor);	40✔
498	tsdb->pScanMonitor = NULL;	40✔
499	}
500	}	40✔

taosdata / TDengine / #4986

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