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

taosdata / TDengine / #3561

19 Dec 2024 03:15AM UTC coverage: 58.812% (-1.3%) from 60.124%
#3561

push

travis-ci

web-flow
Merge pull request #29213 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

130770 of 287658 branches covered (45.46%)

Branch coverage included in aggregate %.

32 of 78 new or added lines in 4 files covered. (41.03%)

7347 existing lines in 166 files now uncovered.

205356 of 283866 relevant lines covered (72.34%)

7187865.64 hits per line

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

50.54
/source/dnode/vnode/src/tsdb/tsdbFS2.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 "tsdbFS2.h"
17
#include "cos.h"
18
#include "tsdbUpgrade.h"
19
#include "vnd.h"
20

21
#define BLOCK_COMMIT_FACTOR 3
22

23
typedef struct STFileHashEntry {
24
  struct STFileHashEntry *next;
25
  char                    fname[TSDB_FILENAME_LEN];
26
} STFileHashEntry;
27

28
typedef struct {
29
  int32_t           numFile;
30
  int32_t           numBucket;
31
  STFileHashEntry **buckets;
32
} STFileHash;
33

34
static const char *gCurrentFname[] = {
35
    [TSDB_FCURRENT] = "current.json",
36
    [TSDB_FCURRENT_C] = "current.c.json",
37
    [TSDB_FCURRENT_M] = "current.m.json",
38
};
39

40
static int32_t create_fs(STsdb *pTsdb, STFileSystem **fs) {
9,579✔
41
  fs[0] = taosMemoryCalloc(1, sizeof(*fs[0]));
9,579!
42
  if (fs[0] == NULL) {
9,628!
43
    return terrno;
×
44
  }
45

46
  fs[0]->tsdb = pTsdb;
9,628✔
47
  int32_t code = tsem_init(&fs[0]->canEdit, 0, 1);
9,628✔
48
  if (code) {
9,627✔
49
    taosMemoryFree(fs[0]);
1!
50
    return code;
×
51
  }
52

53
  fs[0]->fsstate = TSDB_FS_STATE_NORMAL;
9,626✔
54
  fs[0]->neid = 0;
9,626✔
55
  TARRAY2_INIT(fs[0]->fSetArr);
9,626✔
56
  TARRAY2_INIT(fs[0]->fSetArrTmp);
9,626✔
57

58
  return 0;
9,626✔
59
}
60

61
static void destroy_fs(STFileSystem **fs) {
9,631✔
62
  if (fs[0] == NULL) return;
9,631!
63

64
  TARRAY2_DESTROY(fs[0]->fSetArr, NULL);
9,631!
65
  TARRAY2_DESTROY(fs[0]->fSetArrTmp, NULL);
9,631!
66
  if (tsem_destroy(&fs[0]->canEdit) != 0) {
9,631!
67
    tsdbError("failed to destroy semaphore");
×
68
  }
69
  taosMemoryFree(fs[0]);
9,631!
70
  fs[0] = NULL;
9,631✔
71
}
72

73
void current_fname(STsdb *pTsdb, char *fname, EFCurrentT ftype) {
46,761✔
74
  int32_t offset = 0;
46,761✔
75

76
  vnodeGetPrimaryDir(pTsdb->path, pTsdb->pVnode->diskPrimary, pTsdb->pVnode->pTfs, fname, TSDB_FILENAME_LEN);
46,761✔
77
  offset = strlen(fname);
46,787✔
78
  snprintf(fname + offset, TSDB_FILENAME_LEN - offset - 1, "%s%s", TD_DIRSEP, gCurrentFname[ftype]);
46,787✔
79
}
46,787✔
80

81
static int32_t save_json(const cJSON *json, const char *fname) {
13,123✔
82
  int32_t   code = 0;
13,123✔
83
  int32_t   lino;
84
  char     *data = NULL;
13,123✔
85
  TdFilePtr fp = NULL;
13,123✔
86

87
  data = cJSON_PrintUnformatted(json);
13,123✔
88
  if (data == NULL) {
13,121!
89
    TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
×
90
  }
91

92
  fp = taosOpenFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
13,121✔
93
  if (fp == NULL) {
13,122!
94
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
95
  }
96

97
  if (taosWriteFile(fp, data, strlen(data)) < 0) {
13,122!
98
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
99
  }
100

101
  if (taosFsyncFile(fp) < 0) {
13,123!
102
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
103
  }
104

105
_exit:
13,123✔
106
  if (code) {
13,123!
107
    tsdbError("%s failed at %s:%d since %s", __func__, fname, __LINE__, tstrerror(code));
×
108
  }
109
  taosMemoryFree(data);
13,123!
110
  taosCloseFileWithLog(&fp);
13,123!
111
  return code;
13,123✔
112
}
113

114
static int32_t load_json(const char *fname, cJSON **json) {
1,845✔
115
  int32_t code = 0;
1,845✔
116
  int32_t lino;
117
  char   *data = NULL;
1,845✔
118

119
  TdFilePtr fp = taosOpenFile(fname, TD_FILE_READ);
1,845✔
120
  if (fp == NULL) {
1,849!
121
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
122
  }
123

124
  int64_t size;
125
  code = taosFStatFile(fp, &size, NULL);
1,849✔
126
  if (code != 0) {
1,855!
127
    TSDB_CHECK_CODE(code, lino, _exit);
×
128
  }
129

130
  data = taosMemoryMalloc(size + 1);
1,855!
131
  if (data == NULL) {
1,857!
132
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
133
  }
134

135
  if (taosReadFile(fp, data, size) < 0) {
1,857!
136
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
137
  }
138
  data[size] = '\0';
1,851✔
139

140
  json[0] = cJSON_Parse(data);
1,851✔
141
  if (json[0] == NULL) {
1,857!
142
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
143
  }
144

145
_exit:
1,857✔
146
  if (code) {
1,857!
147
    tsdbError("%s failed at %s:%d since %s", __func__, fname, __LINE__, tstrerror(code));
×
148
    json[0] = NULL;
×
149
  }
150
  taosCloseFileWithLog(&fp);
1,857!
151
  taosMemoryFree(data);
1,858!
152
  return code;
1,857✔
153
}
154

155
int32_t save_fs(const TFileSetArray *arr, const char *fname) {
13,123✔
156
  int32_t code = 0;
13,123✔
157
  int32_t lino = 0;
13,123✔
158

159
  cJSON *json = cJSON_CreateObject();
13,123✔
160
  if (json == NULL) {
13,121!
161
    TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
×
162
  }
163

164
  // fmtv
165
  if (cJSON_AddNumberToObject(json, "fmtv", 1) == NULL) {
13,121!
166
    TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
×
167
  }
168

169
  // fset
170
  cJSON *ajson = cJSON_AddArrayToObject(json, "fset");
13,121✔
171
  if (!ajson) {
13,123!
172
    TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
×
173
  }
174
  const STFileSet *fset;
175
  TARRAY2_FOREACH(arr, fset) {
145,260✔
176
    cJSON *item = cJSON_CreateObject();
132,137✔
177
    if (!item) {
132,137!
178
      TSDB_CHECK_CODE(code = TSDB_CODE_OUT_OF_MEMORY, lino, _exit);
×
179
    }
180
    (void)cJSON_AddItemToArray(ajson, item);
132,137✔
181

182
    code = tsdbTFileSetToJson(fset, item);
132,135✔
183
    TSDB_CHECK_CODE(code, lino, _exit);
132,137!
184
  }
185

186
  code = save_json(json, fname);
13,123✔
187
  TSDB_CHECK_CODE(code, lino, _exit);
13,123!
188

189
_exit:
13,123✔
190
  if (code) {
13,123!
191
    tsdbError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
×
192
  }
193
  cJSON_Delete(json);
13,123✔
194
  return code;
13,123✔
195
}
196

197
static int32_t load_fs(STsdb *pTsdb, const char *fname, TFileSetArray *arr) {
1,853✔
198
  int32_t code = 0;
1,853✔
199
  int32_t lino = 0;
1,853✔
200

201
  TARRAY2_CLEAR(arr, tsdbTFileSetClear);
1,853!
202

203
  // load json
204
  cJSON *json = NULL;
1,853✔
205
  code = load_json(fname, &json);
1,853✔
206
  TSDB_CHECK_CODE(code, lino, _exit);
1,857!
207

208
  // parse json
209
  const cJSON *item1;
210

211
  /* fmtv */
212
  item1 = cJSON_GetObjectItem(json, "fmtv");
1,857✔
213
  if (cJSON_IsNumber(item1)) {
1,858!
214
    if (item1->valuedouble != 1) {
1,856!
215
      TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
216
    }
217
  } else {
218
    TSDB_CHECK_CODE(code = TSDB_CODE_FILE_CORRUPTED, lino, _exit);
×
219
  }
220

221
  /* fset */
222
  item1 = cJSON_GetObjectItem(json, "fset");
1,856✔
223
  if (cJSON_IsArray(item1)) {
1,858!
224
    const cJSON *item2;
225
    cJSON_ArrayForEach(item2, item1) {
3,406!
226
      STFileSet *fset;
227
      code = tsdbJsonToTFileSet(pTsdb, item2, &fset);
1,548✔
228
      TSDB_CHECK_CODE(code, lino, _exit);
1,548!
229

230
      code = TARRAY2_APPEND(arr, fset);
1,548✔
231
      TSDB_CHECK_CODE(code, lino, _exit);
1,549!
232
    }
233
    TARRAY2_SORT(arr, tsdbTFileSetCmprFn);
1,858✔
234
  } else {
235
    code = TSDB_CODE_FILE_CORRUPTED;
×
236
    TSDB_CHECK_CODE(code, lino, _exit);
×
237
  }
238

239
_exit:
×
240
  if (code) {
1,858!
241
    tsdbError("%s failed at %sP%d since %s, fname:%s", __func__, __FILE__, lino, tstrerror(code), fname);
×
242
  }
243
  if (json) {
1,855!
244
    cJSON_Delete(json);
1,855✔
245
  }
246
  return code;
1,858✔
247
}
248

249
static int32_t apply_commit(STFileSystem *fs) {
5,350✔
250
  int32_t        code = 0;
5,350✔
251
  int32_t        lino;
252
  TFileSetArray *fsetArray1 = fs->fSetArr;
5,350✔
253
  TFileSetArray *fsetArray2 = fs->fSetArrTmp;
5,350✔
254
  int32_t        i1 = 0, i2 = 0;
5,350✔
255

256
  while (i1 < TARRAY2_SIZE(fsetArray1) || i2 < TARRAY2_SIZE(fsetArray2)) {
137,444✔
257
    STFileSet *fset1 = i1 < TARRAY2_SIZE(fsetArray1) ? TARRAY2_GET(fsetArray1, i1) : NULL;
132,096✔
258
    STFileSet *fset2 = i2 < TARRAY2_SIZE(fsetArray2) ? TARRAY2_GET(fsetArray2, i2) : NULL;
132,096!
259

260
    if (fset1 && fset2) {
132,096!
261
      if (fset1->fid < fset2->fid) {
5,358!
262
        // delete fset1
263
        tsdbTFileSetRemove(fset1);
×
264
        i1++;
×
265
      } else if (fset1->fid > fset2->fid) {
5,358✔
266
        // create new file set with fid of fset2->fid
267
        code = tsdbTFileSetInitCopy(fs->tsdb, fset2, &fset1);
26✔
268
        TSDB_CHECK_CODE(code, lino, _exit);
26!
269
        code = TARRAY2_SORT_INSERT(fsetArray1, fset1, tsdbTFileSetCmprFn);
26✔
270
        TSDB_CHECK_CODE(code, lino, _exit);
26!
271
        i1++;
26✔
272
        i2++;
26✔
273
      } else {
274
        // edit
275
        code = tsdbTFileSetApplyEdit(fs->tsdb, fset2, fset1);
5,332✔
276
        TSDB_CHECK_CODE(code, lino, _exit);
5,332!
277
        i1++;
5,332✔
278
        i2++;
5,332✔
279
      }
280
    } else if (fset1) {
126,738!
281
      // delete fset1
282
      tsdbTFileSetRemove(fset1);
×
283
      i1++;
×
284
    } else {
285
      // create new file set with fid of fset2->fid
286
      code = tsdbTFileSetInitCopy(fs->tsdb, fset2, &fset1);
126,738✔
287
      TSDB_CHECK_CODE(code, lino, _exit);
126,779!
288
      code = TARRAY2_SORT_INSERT(fsetArray1, fset1, tsdbTFileSetCmprFn);
126,736✔
289
      TSDB_CHECK_CODE(code, lino, _exit);
126,736!
290
      i1++;
126,736✔
291
      i2++;
126,736✔
292
    }
293
  }
294

295
_exit:
5,348✔
296
  if (code) {
5,348!
297
    TSDB_ERROR_LOG(TD_VID(fs->tsdb->pVnode), lino, code);
×
298
  }
299
  return code;
5,350✔
300
}
301

302
static int32_t commit_edit(STFileSystem *fs) {
5,350✔
303
  char current[TSDB_FILENAME_LEN];
304
  char current_t[TSDB_FILENAME_LEN];
305

306
  current_fname(fs->tsdb, current, TSDB_FCURRENT);
5,350✔
307
  if (fs->etype == TSDB_FEDIT_COMMIT) {
5,350✔
308
    current_fname(fs->tsdb, current_t, TSDB_FCURRENT_C);
4,286✔
309
  } else {
310
    current_fname(fs->tsdb, current_t, TSDB_FCURRENT_M);
1,064✔
311
  }
312

313
  int32_t code;
314
  int32_t lino;
315
  TSDB_CHECK_CODE(taosRenameFile(current_t, current), lino, _exit);
5,350!
316

317
  code = apply_commit(fs);
5,350✔
318
  TSDB_CHECK_CODE(code, lino, _exit);
5,350!
319

320
_exit:
5,350✔
321
  if (code) {
5,350!
322
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(fs->tsdb->pVnode), __func__, __FILE__, lino,
×
323
              tstrerror(code));
324
  } else {
325
    tsdbInfo("vgId:%d %s success, etype:%d", TD_VID(fs->tsdb->pVnode), __func__, fs->etype);
5,350!
326
  }
327
  return code;
5,350✔
328
}
329

330
// static int32_t
331
static int32_t tsdbFSDoSanAndFix(STFileSystem *fs);
332
static int32_t apply_abort(STFileSystem *fs) { return tsdbFSDoSanAndFix(fs); }
×
333

334
static int32_t abort_edit(STFileSystem *fs) {
×
335
  char fname[TSDB_FILENAME_LEN];
336

337
  if (fs->etype == TSDB_FEDIT_COMMIT) {
×
338
    current_fname(fs->tsdb, fname, TSDB_FCURRENT_C);
×
339
  } else {
340
    current_fname(fs->tsdb, fname, TSDB_FCURRENT_M);
×
341
  }
342

343
  int32_t code;
344
  int32_t lino;
345
  if ((code = taosRemoveFile(fname))) {
×
346
    code = TAOS_SYSTEM_ERROR(code);
×
347
    TSDB_CHECK_CODE(code, lino, _exit);
×
348
  }
349

350
  code = apply_abort(fs);
×
351
  TSDB_CHECK_CODE(code, lino, _exit);
×
352

353
_exit:
×
354
  if (code) {
×
355
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(fs->tsdb->pVnode), __func__, __FILE__, lino,
×
356
              tstrerror(code));
357
  } else {
358
    tsdbInfo("vgId:%d %s success, etype:%d", TD_VID(fs->tsdb->pVnode), __func__, fs->etype);
×
359
  }
360
  return code;
×
361
}
362

363
static int32_t tsdbFSDoScanAndFixFile(STFileSystem *fs, const STFileObj *fobj) {
2,784✔
364
  int32_t code = 0;
2,784✔
365
  int32_t lino = 0;
2,784✔
366

367
  // check file existence
368
  if (!taosCheckExistFile(fobj->fname)) {
2,784!
369
    bool found = false;
×
370

371
    if (tsS3Enabled && fobj->f->lcn > 1) {
×
372
      char fname1[TSDB_FILENAME_LEN];
373
      tsdbTFileLastChunkName(fs->tsdb, fobj->f, fname1);
×
374
      if (!taosCheckExistFile(fname1)) {
×
375
        code = TSDB_CODE_FILE_CORRUPTED;
×
376
        tsdbError("vgId:%d %s failed since file:%s does not exist", TD_VID(fs->tsdb->pVnode), __func__, fname1);
×
377
        return code;
×
378
      }
379

380
      found = true;
×
381
    }
382

383
    if (!found) {
×
384
      code = TSDB_CODE_FILE_CORRUPTED;
×
385
      tsdbError("vgId:%d %s failed since file:%s does not exist", TD_VID(fs->tsdb->pVnode), __func__, fobj->fname);
×
386
      return code;
×
387
    }
388
  }
389

390
  return 0;
2,792✔
391
}
392

393
static void tsdbFSDestroyFileObjHash(STFileHash *hash);
394

395
static int32_t tsdbFSAddEntryToFileObjHash(STFileHash *hash, const char *fname) {
4,646✔
396
  STFileHashEntry *entry = taosMemoryMalloc(sizeof(*entry));
4,646!
397
  if (entry == NULL) return terrno;
4,649!
398

399
  tstrncpy(entry->fname, fname, TSDB_FILENAME_LEN);
4,649✔
400

401
  uint32_t idx = MurmurHash3_32(fname, strlen(fname)) % hash->numBucket;
4,649✔
402

403
  entry->next = hash->buckets[idx];
4,650✔
404
  hash->buckets[idx] = entry;
4,650✔
405
  hash->numFile++;
4,650✔
406

407
  return 0;
4,650✔
408
}
409

410
static int32_t tsdbFSCreateFileObjHash(STFileSystem *fs, STFileHash *hash) {
1,851✔
411
  int32_t code = 0;
1,851✔
412
  int32_t lino;
413
  char    fname[TSDB_FILENAME_LEN];
414

415
  // init hash table
416
  hash->numFile = 0;
1,851✔
417
  hash->numBucket = 4096;
1,851✔
418
  hash->buckets = taosMemoryCalloc(hash->numBucket, sizeof(STFileHashEntry *));
1,851!
419
  if (hash->buckets == NULL) {
1,858!
420
    TSDB_CHECK_CODE(code = terrno, lino, _exit);
×
421
  }
422

423
  // vnode.json
424
  current_fname(fs->tsdb, fname, TSDB_FCURRENT);
1,858✔
425
  code = tsdbFSAddEntryToFileObjHash(hash, fname);
1,857✔
426
  TSDB_CHECK_CODE(code, lino, _exit);
1,858!
427

428
  // other
429
  STFileSet *fset = NULL;
1,858✔
430
  TARRAY2_FOREACH(fs->fSetArr, fset) {
3,404✔
431
    // data file
432
    for (int32_t i = 0; i < TSDB_FTYPE_MAX; i++) {
7,738✔
433
      if (fset->farr[i] != NULL) {
6,191✔
434
        code = tsdbFSAddEntryToFileObjHash(hash, fset->farr[i]->fname);
1,142✔
435
        TSDB_CHECK_CODE(code, lino, _exit);
1,142!
436

437
        if (TSDB_FTYPE_DATA == i && fset->farr[i]->f->lcn > 0) {
1,142!
438
          STFileObj *fobj = fset->farr[i];
×
439
          int32_t    lcn = fobj->f->lcn;
×
440
          char       lcn_name[TSDB_FILENAME_LEN];
441

442
          snprintf(lcn_name, TSDB_FQDN_LEN, "%s", fobj->fname);
×
443
          char *dot = strrchr(lcn_name, '.');
×
444
          if (dot) {
×
445
            snprintf(dot + 1, TSDB_FQDN_LEN - (dot + 1 - lcn_name), "%d.data", lcn);
×
446

447
            code = tsdbFSAddEntryToFileObjHash(hash, lcn_name);
×
448
            TSDB_CHECK_CODE(code, lino, _exit);
×
449
          }
450
        }
451
      }
452
    }
453

454
    // stt file
455
    SSttLvl *lvl = NULL;
1,547✔
456
    TARRAY2_FOREACH(fset->lvlArr, lvl) {
3,060✔
457
      STFileObj *fobj;
458
      TARRAY2_FOREACH(lvl->fobjArr, fobj) {
3,163✔
459
        code = tsdbFSAddEntryToFileObjHash(hash, fobj->fname);
1,650✔
460
        TSDB_CHECK_CODE(code, lino, _exit);
1,649!
461
      }
462
    }
463
  }
464

465
_exit:
1,857✔
466
  if (code) {
1,857!
467
    TSDB_ERROR_LOG(TD_VID(fs->tsdb->pVnode), lino, code);
×
468
    tsdbFSDestroyFileObjHash(hash);
×
469
  }
470
  return code;
1,858✔
471
}
472

473
static const STFileHashEntry *tsdbFSGetFileObjHashEntry(STFileHash *hash, const char *fname) {
4,649✔
474
  uint32_t idx = MurmurHash3_32(fname, strlen(fname)) % hash->numBucket;
4,649✔
475

476
  STFileHashEntry *entry = hash->buckets[idx];
4,649✔
477
  while (entry) {
4,656!
478
    if (strcmp(entry->fname, fname) == 0) {
4,656✔
479
      return entry;
4,649✔
480
    }
481
    entry = entry->next;
7✔
482
  }
483

UNCOV
484
  return NULL;
×
485
}
486

487
static void tsdbFSDestroyFileObjHash(STFileHash *hash) {
1,858✔
488
  for (int32_t i = 0; i < hash->numBucket; i++) {
7,534,200✔
489
    STFileHashEntry *entry = hash->buckets[i];
7,532,342✔
490
    while (entry) {
7,536,992✔
491
      STFileHashEntry *next = entry->next;
4,650✔
492
      taosMemoryFree(entry);
4,650!
493
      entry = next;
4,650✔
494
    }
495
  }
496
  taosMemoryFree(hash->buckets);
1,858!
497
  memset(hash, 0, sizeof(*hash));
1,858✔
498
}
1,858✔
499

500
static int32_t tsdbFSDoSanAndFix(STFileSystem *fs) {
1,849✔
501
  int32_t code = 0;
1,849✔
502
  int32_t lino = 0;
1,849✔
503
  int32_t corrupt = false;
1,849✔
504

505
  {  // scan each file
506
    STFileSet *fset = NULL;
1,849✔
507
    TARRAY2_FOREACH(fs->fSetArr, fset) {
3,392✔
508
      // data file
509
      for (int32_t ftype = 0; ftype < TSDB_FTYPE_MAX; ftype++) {
7,717✔
510
        if (fset->farr[ftype] == NULL) continue;
6,175✔
511
        STFileObj *fobj = fset->farr[ftype];
1,142✔
512
        code = tsdbFSDoScanAndFixFile(fs, fobj);
1,142✔
513
        if (code) {
1,142!
514
          fset->maxVerValid = (fobj->f->minVer <= fobj->f->maxVer) ? TMIN(fset->maxVerValid, fobj->f->minVer - 1) : -1;
×
515
          corrupt = true;
×
516
        }
517
      }
518

519
      // stt file
520
      SSttLvl *lvl;
521
      TARRAY2_FOREACH(fset->lvlArr, lvl) {
3,051✔
522
        STFileObj *fobj;
523
        TARRAY2_FOREACH(lvl->fobjArr, fobj) {
3,154✔
524
          code = tsdbFSDoScanAndFixFile(fs, fobj);
1,645✔
525
          if (code) {
1,646!
526
            fset->maxVerValid =
×
527
                (fobj->f->minVer <= fobj->f->maxVer) ? TMIN(fset->maxVerValid, fobj->f->minVer - 1) : -1;
×
528
            corrupt = true;
×
529
          }
530
        }
531
      }
532
    }
533
  }
534

535
  if (corrupt) {
1,850!
536
    tsdbError("vgId:%d, not to clear dangling files due to fset incompleteness", TD_VID(fs->tsdb->pVnode));
×
537
    fs->fsstate = TSDB_FS_STATE_INCOMPLETE;
×
538
    code = 0;
×
539
    goto _exit;
×
540
  }
541

542
  {  // clear unreferenced files
543
    STfsDir *dir = NULL;
1,850✔
544
    TAOS_CHECK_GOTO(tfsOpendir(fs->tsdb->pVnode->pTfs, fs->tsdb->path, &dir), &lino, _exit);
1,850!
545

546
    STFileHash fobjHash = {0};
1,855✔
547
    code = tsdbFSCreateFileObjHash(fs, &fobjHash);
1,855✔
548
    if (code) goto _close_dir;
1,858!
549

550
    for (const STfsFile *file = NULL; (file = tfsReaddir(dir)) != NULL;) {
8,365✔
551
      if (taosIsDir(file->aname)) continue;
6,506✔
552

553
      if (tsdbFSGetFileObjHashEntry(&fobjHash, file->aname) == NULL) {
4,649!
554
        tsdbRemoveFile(file->aname);
×
555
      }
556
    }
557

558
    tsdbFSDestroyFileObjHash(&fobjHash);
1,858✔
559

560
  _close_dir:
1,858✔
561
    tfsClosedir(dir);
1,858✔
562
  }
563

564
_exit:
1,858✔
565
  if (code) {
1,858!
566
    TSDB_ERROR_LOG(TD_VID(fs->tsdb->pVnode), lino, code);
×
567
  }
568
  return code;
1,858✔
569
}
570

571
static int32_t tsdbFSScanAndFix(STFileSystem *fs) {
1,849✔
572
  fs->neid = 0;
1,849✔
573

574
  // get max commit id
575
  const STFileSet *fset;
576
  TARRAY2_FOREACH(fs->fSetArr, fset) { fs->neid = TMAX(fs->neid, tsdbTFileSetMaxCid(fset)); }
3,397✔
577

578
  // scan and fix
579
  int32_t code = 0;
1,850✔
580
  int32_t lino = 0;
1,850✔
581

582
  code = tsdbFSDoSanAndFix(fs);
1,850✔
583
  TSDB_CHECK_CODE(code, lino, _exit);
1,858!
584

585
_exit:
1,858✔
586
  if (code) {
1,858!
587
    TSDB_ERROR_LOG(TD_VID(fs->tsdb->pVnode), lino, code);
×
588
  }
589
  return code;
1,858✔
590
}
591

592
static int32_t tsdbFSDupState(STFileSystem *fs) {
7,207✔
593
  int32_t code;
594

595
  const TFileSetArray *src = fs->fSetArr;
7,207✔
596
  TFileSetArray       *dst = fs->fSetArrTmp;
7,207✔
597

598
  TARRAY2_CLEAR(dst, tsdbTFileSetClear);
12,539✔
599

600
  const STFileSet *fset1;
601
  TARRAY2_FOREACH(src, fset1) {
14,084✔
602
    STFileSet *fset2;
603
    code = tsdbTFileSetInitCopy(fs->tsdb, fset1, &fset2);
6,880✔
604
    if (code) return code;
6,875!
605
    code = TARRAY2_APPEND(dst, fset2);
6,875✔
606
    if (code) return code;
6,877!
607
  }
608

609
  return 0;
7,204✔
610
}
611

612
static int32_t open_fs(STFileSystem *fs, int8_t rollback) {
9,619✔
613
  int32_t code = 0;
9,619✔
614
  int32_t lino = 0;
9,619✔
615
  STsdb  *pTsdb = fs->tsdb;
9,619✔
616

617
  char fCurrent[TSDB_FILENAME_LEN];
618
  char cCurrent[TSDB_FILENAME_LEN];
619
  char mCurrent[TSDB_FILENAME_LEN];
620

621
  current_fname(pTsdb, fCurrent, TSDB_FCURRENT);
9,619✔
622
  current_fname(pTsdb, cCurrent, TSDB_FCURRENT_C);
9,615✔
623
  current_fname(pTsdb, mCurrent, TSDB_FCURRENT_M);
9,628✔
624

625
  if (taosCheckExistFile(fCurrent)) {  // current.json exists
9,631✔
626
    code = load_fs(pTsdb, fCurrent, fs->fSetArr);
1,857✔
627
    TSDB_CHECK_CODE(code, lino, _exit);
1,855!
628

629
    if (taosCheckExistFile(cCurrent)) {
1,855!
630
      // current.c.json exists
631

632
      fs->etype = TSDB_FEDIT_COMMIT;
×
633
      if (rollback) {
×
634
        code = abort_edit(fs);
×
635
        TSDB_CHECK_CODE(code, lino, _exit);
×
636
      } else {
637
        code = load_fs(pTsdb, cCurrent, fs->fSetArrTmp);
×
638
        TSDB_CHECK_CODE(code, lino, _exit);
×
639

640
        code = commit_edit(fs);
×
641
        TSDB_CHECK_CODE(code, lino, _exit);
×
642
      }
643
    } else if (taosCheckExistFile(mCurrent)) {
1,856!
644
      // current.m.json exists
645
      fs->etype = TSDB_FEDIT_MERGE;
×
646
      code = abort_edit(fs);
×
647
      TSDB_CHECK_CODE(code, lino, _exit);
×
648
    }
649

650
    code = tsdbFSDupState(fs);
1,857✔
651
    TSDB_CHECK_CODE(code, lino, _exit);
1,852!
652

653
    code = tsdbFSScanAndFix(fs);
1,852✔
654
    TSDB_CHECK_CODE(code, lino, _exit);
1,858!
655
  } else {
656
    code = save_fs(fs->fSetArr, fCurrent);
7,773✔
657
    TSDB_CHECK_CODE(code, lino, _exit);
7,773!
658
  }
659

660
_exit:
7,773✔
661
  if (code) {
9,631!
662
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
×
663
  } else {
664
    tsdbInfo("vgId:%d %s success", TD_VID(pTsdb->pVnode), __func__);
9,631✔
665
  }
666
  return code;
9,631✔
667
}
668

669
static void close_file_system(STFileSystem *fs) {
9,631✔
670
  TARRAY2_CLEAR(fs->fSetArr, tsdbTFileSetClear);
137,983✔
671
  TARRAY2_CLEAR(fs->fSetArrTmp, tsdbTFileSetClear);
137,986✔
672
}
9,632✔
673

674
static int32_t fset_cmpr_fn(const struct STFileSet *pSet1, const struct STFileSet *pSet2) {
×
675
  if (pSet1->fid < pSet2->fid) {
×
676
    return -1;
×
677
  } else if (pSet1->fid > pSet2->fid) {
×
678
    return 1;
×
679
  }
680
  return 0;
×
681
}
682

683
static int32_t edit_fs(STFileSystem *fs, const TFileOpArray *opArray) {
5,350✔
684
  int32_t code = 0;
5,350✔
685
  int32_t lino = 0;
5,350✔
686

687
  code = tsdbFSDupState(fs);
5,350✔
688
  if (code) return code;
5,350!
689

690
  TFileSetArray  *fsetArray = fs->fSetArrTmp;
5,350✔
691
  STFileSet      *fset = NULL;
5,350✔
692
  const STFileOp *op;
693
  TARRAY2_FOREACH_PTR(opArray, op) {
141,471✔
694
    if (!fset || fset->fid != op->fid) {
136,121✔
695
      STFileSet tfset = {.fid = op->fid};
130,084✔
696
      fset = &tfset;
130,084✔
697
      STFileSet **fsetPtr = TARRAY2_SEARCH(fsetArray, &fset, tsdbTFileSetCmprFn, TD_EQ);
130,077✔
698
      fset = (fsetPtr == NULL) ? NULL : *fsetPtr;
130,077✔
699

700
      if (!fset) {
130,077✔
701
        code = tsdbTFileSetInit(op->fid, &fset);
126,800✔
702
        TSDB_CHECK_CODE(code, lino, _exit);
126,794!
703

704
        code = TARRAY2_SORT_INSERT(fsetArray, fset, tsdbTFileSetCmprFn);
126,797✔
705
        TSDB_CHECK_CODE(code, lino, _exit);
126,797!
706
      }
707
    }
708

709
    code = tsdbTFileSetEdit(fs->tsdb, fset, op);
136,111✔
710
    TSDB_CHECK_CODE(code, lino, _exit);
136,121!
711
  }
712

713
  // remove empty empty stt level and empty file set
714
  int32_t i = 0;
5,350✔
715
  while (i < TARRAY2_SIZE(fsetArray)) {
137,488✔
716
    fset = TARRAY2_GET(fsetArray, i);
132,138✔
717

718
    SSttLvl *lvl;
719
    int32_t  j = 0;
132,138✔
720
    while (j < TARRAY2_SIZE(fset->lvlArr)) {
265,907✔
721
      lvl = TARRAY2_GET(fset->lvlArr, j);
133,769✔
722

723
      if (TARRAY2_SIZE(lvl->fobjArr) == 0) {
133,769✔
724
        TARRAY2_REMOVE(fset->lvlArr, j, tsdbSttLvlClear);
1,420!
725
      } else {
726
        j++;
132,349✔
727
      }
728
    }
729

730
    if (tsdbTFileSetIsEmpty(fset)) {
132,138!
731
      TARRAY2_REMOVE(fsetArray, i, tsdbTFileSetClear);
×
732
    } else {
733
      i++;
132,138✔
734
    }
735
  }
736

737
_exit:
5,350✔
738
  if (code) {
5,350!
739
    TSDB_ERROR_LOG(TD_VID(fs->tsdb->pVnode), lino, code);
×
740
  }
741
  return code;
5,350✔
742
}
743

744
// return error code
745
int32_t tsdbOpenFS(STsdb *pTsdb, STFileSystem **fs, int8_t rollback) {
9,599✔
746
  int32_t code;
747
  int32_t lino;
748

749
  code = tsdbCheckAndUpgradeFileSystem(pTsdb, rollback);
9,599✔
750
  TSDB_CHECK_CODE(code, lino, _exit);
9,624!
751

752
  code = create_fs(pTsdb, fs);
9,624✔
753
  TSDB_CHECK_CODE(code, lino, _exit);
9,627!
754

755
  code = open_fs(fs[0], rollback);
9,627✔
756
  TSDB_CHECK_CODE(code, lino, _exit);
9,631!
757

758
_exit:
9,631✔
759
  if (code) {
9,631!
760
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, lino, tstrerror(code));
×
761
    destroy_fs(fs);
×
762
  } else {
763
    tsdbInfo("vgId:%d %s success", TD_VID(pTsdb->pVnode), __func__);
9,631✔
764
  }
765
  return code;
9,631✔
766
}
767

768
static void tsdbFSSetBlockCommit(STFileSet *fset, bool block);
769
extern void tsdbStopAllCompTask(STsdb *tsdb);
770

771
int32_t tsdbDisableAndCancelAllBgTask(STsdb *pTsdb) {
9,696✔
772
  STFileSystem *fs = pTsdb->pFS;
9,696✔
773
  SArray       *channelArray = taosArrayInit(0, sizeof(SVAChannelID));
9,696✔
774
  if (channelArray == NULL) {
9,696!
775
    return terrno;
×
776
  }
777

778
  (void)taosThreadMutexLock(&pTsdb->mutex);
9,696✔
779

780
  // disable
781
  pTsdb->bgTaskDisabled = true;
9,696✔
782

783
  // collect channel
784
  STFileSet *fset;
785
  TARRAY2_FOREACH(fs->fSetArr, fset) {
138,052✔
786
    if (fset->channelOpened) {
128,356✔
787
      if (taosArrayPush(channelArray, &fset->channel) == NULL) {
970!
788
        taosArrayDestroy(channelArray);
×
789
        (void)taosThreadMutexUnlock(&pTsdb->mutex);
×
790
        return terrno;
×
791
      }
792
      fset->channel = (SVAChannelID){0};
485✔
793
      fset->mergeScheduled = false;
485✔
794
      tsdbFSSetBlockCommit(fset, false);
485✔
795
      fset->channelOpened = false;
485✔
796
    }
797
  }
798

799
  (void)taosThreadMutexUnlock(&pTsdb->mutex);
9,696✔
800

801
  // destroy all channels
802
  for (int32_t i = 0; i < taosArrayGetSize(channelArray); i++) {
10,181✔
803
    SVAChannelID *channel = taosArrayGet(channelArray, i);
485✔
804
    int32_t       code = vnodeAChannelDestroy(channel, true);
485✔
805
    if (code) {
485!
806
      tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tstrerror(code));
×
807
    }
808
  }
809
  taosArrayDestroy(channelArray);
9,696✔
810

811
#ifdef TD_ENTERPRISE
812
  tsdbStopAllCompTask(pTsdb);
9,696✔
813
#endif
814
  return 0;
9,696✔
815
}
816

817
void tsdbEnableBgTask(STsdb *pTsdb) {
65✔
818
  (void)taosThreadMutexLock(&pTsdb->mutex);
65✔
819
  pTsdb->bgTaskDisabled = false;
65✔
820
  (void)taosThreadMutexUnlock(&pTsdb->mutex);
65✔
821
}
65✔
822

823
void tsdbCloseFS(STFileSystem **fs) {
9,631✔
824
  if (fs[0] == NULL) return;
9,631!
825

826
  int32_t code = tsdbDisableAndCancelAllBgTask((*fs)->tsdb);
9,631✔
827
  if (code) {
9,631!
828
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID((*fs)->tsdb->pVnode), __func__, __LINE__,
×
829
              tstrerror(code));
830
  }
831
  close_file_system(fs[0]);
9,631✔
832
  destroy_fs(fs);
9,631✔
833
  return;
9,631✔
834
}
835

836
int64_t tsdbFSAllocEid(STFileSystem *fs) {
5,428✔
837
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
5,428✔
838
  int64_t cid = ++fs->neid;
5,428✔
839
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);
5,428✔
840
  return cid;
5,428✔
841
}
842

843
void tsdbFSUpdateEid(STFileSystem *fs, int64_t cid) {
337✔
844
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
337✔
845
  fs->neid = TMAX(fs->neid, cid);
337✔
846
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);
337✔
847
}
337✔
848

849
int32_t tsdbFSEditBegin(STFileSystem *fs, const TFileOpArray *opArray, EFEditT etype) {
5,350✔
850
  int32_t code = 0;
5,350✔
851
  int32_t lino;
852
  char    current_t[TSDB_FILENAME_LEN];
853

854
  if (etype == TSDB_FEDIT_COMMIT) {
5,350✔
855
    current_fname(fs->tsdb, current_t, TSDB_FCURRENT_C);
4,286✔
856
  } else {
857
    current_fname(fs->tsdb, current_t, TSDB_FCURRENT_M);
1,064✔
858
  }
859

860
  if (tsem_wait(&fs->canEdit) != 0) {
5,350!
861
    tsdbError("vgId:%d failed to wait semaphore", TD_VID(fs->tsdb->pVnode));
×
862
  }
863
  fs->etype = etype;
5,350✔
864

865
  // edit
866
  code = edit_fs(fs, opArray);
5,350✔
867
  TSDB_CHECK_CODE(code, lino, _exit);
5,350!
868

869
  // save fs
870
  code = save_fs(fs->fSetArrTmp, current_t);
5,350✔
871
  TSDB_CHECK_CODE(code, lino, _exit);
5,350!
872

873
_exit:
5,350✔
874
  if (code) {
5,350!
875
    tsdbError("vgId:%d %s failed at line %d since %s, etype:%d", TD_VID(fs->tsdb->pVnode), __func__, lino,
×
876
              tstrerror(code), etype);
877
  } else {
878
    tsdbInfo("vgId:%d %s done, etype:%d", TD_VID(fs->tsdb->pVnode), __func__, etype);
5,350!
879
  }
880
  return code;
5,350✔
881
}
882

883
static void tsdbFSSetBlockCommit(STFileSet *fset, bool block) {
130,394✔
884
  if (block) {
130,394!
UNCOV
885
    fset->blockCommit = true;
×
886
  } else {
887
    fset->blockCommit = false;
130,394✔
888
    if (fset->numWaitCommit > 0) {
130,394!
UNCOV
889
      (void)taosThreadCondSignal(&fset->canCommit);
×
890
    }
891
  }
892
}
130,394✔
893

894
void tsdbFSCheckCommit(STsdb *tsdb, int32_t fid) {
128,919✔
895
  (void)taosThreadMutexLock(&tsdb->mutex);
128,919✔
896
  STFileSet *fset;
897
  tsdbFSGetFSet(tsdb->pFS, fid, &fset);
128,917✔
898
  if (fset) {
128,899✔
899
    while (fset->blockCommit) {
2,216!
UNCOV
900
      fset->numWaitCommit++;
×
UNCOV
901
      (void)taosThreadCondWait(&fset->canCommit, &tsdb->mutex);
×
UNCOV
902
      fset->numWaitCommit--;
×
903
    }
904
  }
905
  (void)taosThreadMutexUnlock(&tsdb->mutex);
128,899✔
906
  return;
128,923✔
907
}
908

909
// IMPORTANT: the caller must hold fs->tsdb->mutex
910
int32_t tsdbFSEditCommit(STFileSystem *fs) {
5,350✔
911
  int32_t code = 0;
5,350✔
912
  int32_t lino = 0;
5,350✔
913

914
  // commit
915
  code = commit_edit(fs);
5,350✔
916
  TSDB_CHECK_CODE(code, lino, _exit);
5,350!
917

918
  // schedule merge
919
  int32_t sttTrigger = fs->tsdb->pVnode->config.sttTrigger;
5,350✔
920
  if (sttTrigger > 1 && !fs->tsdb->bgTaskDisabled) {
5,350✔
921
    STFileSet *fset;
922
    TARRAY2_FOREACH_REVERSE(fs->fSetArr, fset) {
135,125✔
923
      if (TARRAY2_SIZE(fset->lvlArr) == 0) {
129,909✔
924
        tsdbFSSetBlockCommit(fset, false);
466✔
925
        continue;
466✔
926
      }
927

928
      SSttLvl *lvl = TARRAY2_FIRST(fset->lvlArr);
129,443✔
929
      if (lvl->level != 0) {
129,443✔
930
        tsdbFSSetBlockCommit(fset, false);
1,131✔
931
        continue;
1,131✔
932
      }
933

934
      // bool    skipMerge = false;
935
      int32_t numFile = TARRAY2_SIZE(lvl->fobjArr);
128,312✔
936
      if (numFile >= sttTrigger && (!fset->mergeScheduled)) {
128,312✔
937
        code = tsdbTFileSetOpenChannel(fset);
1,165✔
938
        TSDB_CHECK_CODE(code, lino, _exit);
1,165!
939

940
        SMergeArg *arg = taosMemoryMalloc(sizeof(*arg));
1,165!
941
        if (arg == NULL) {
1,165!
942
          code = terrno;
×
943
          TSDB_CHECK_CODE(code, lino, _exit);
×
944
        }
945

946
        arg->tsdb = fs->tsdb;
1,165✔
947
        arg->fid = fset->fid;
1,165✔
948

949
        code = vnodeAsync(&fset->channel, EVA_PRIORITY_HIGH, tsdbMerge, taosAutoMemoryFree, arg, NULL);
1,165✔
950
        TSDB_CHECK_CODE(code, lino, _exit);
1,165!
951
        fset->mergeScheduled = true;
1,165✔
952
      }
953

954
      if (numFile >= sttTrigger * BLOCK_COMMIT_FACTOR) {
128,312!
UNCOV
955
        tsdbFSSetBlockCommit(fset, true);
×
956
      } else {
957
        tsdbFSSetBlockCommit(fset, false);
128,312✔
958
      }
959
    }
960
  }
961

962
_exit:
5,350✔
963
  if (code) {
5,350!
964
    tsdbError("vgId:%d %s failed at line %d since %s", TD_VID(fs->tsdb->pVnode), __func__, lino, tstrerror(code));
×
965
  } else {
966
    tsdbInfo("vgId:%d %s done, etype:%d", TD_VID(fs->tsdb->pVnode), __func__, fs->etype);
5,350!
967
  }
968
  if (tsem_post(&fs->canEdit) != 0) {
5,350!
969
    tsdbError("vgId:%d failed to post semaphore", TD_VID(fs->tsdb->pVnode));
×
970
  }
971
  return code;
5,350✔
972
}
973

974
int32_t tsdbFSEditAbort(STFileSystem *fs) {
×
975
  int32_t code = abort_edit(fs);
×
976
  if (tsem_post(&fs->canEdit) != 0) {
×
977
    tsdbError("vgId:%d failed to post semaphore", TD_VID(fs->tsdb->pVnode));
×
978
  }
979
  return code;
×
980
}
981

982
void tsdbFSGetFSet(STFileSystem *fs, int32_t fid, STFileSet **fset) {
259,395✔
983
  STFileSet   tfset = {.fid = fid};
259,395✔
984
  STFileSet  *pset = &tfset;
259,395✔
985
  STFileSet **fsetPtr = TARRAY2_SEARCH(fs->fSetArr, &pset, tsdbTFileSetCmprFn, TD_EQ);
259,395✔
986
  fset[0] = (fsetPtr == NULL) ? NULL : fsetPtr[0];
259,391✔
987
}
259,391✔
988

989
int32_t tsdbFSCreateCopySnapshot(STFileSystem *fs, TFileSetArray **fsetArr) {
65✔
990
  int32_t    code = 0;
65✔
991
  STFileSet *fset;
992
  STFileSet *fset1;
993

994
  fsetArr[0] = taosMemoryMalloc(sizeof(TFileSetArray));
65!
995
  if (fsetArr[0] == NULL) return terrno;
65!
996

997
  TARRAY2_INIT(fsetArr[0]);
65✔
998

999
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
65✔
1000
  TARRAY2_FOREACH(fs->fSetArr, fset) {
65!
1001
    code = tsdbTFileSetInitCopy(fs->tsdb, fset, &fset1);
×
1002
    if (code) break;
×
1003

1004
    code = TARRAY2_APPEND(fsetArr[0], fset1);
×
1005
    if (code) break;
×
1006
  }
1007
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);
65✔
1008

1009
  if (code) {
65!
1010
    TARRAY2_DESTROY(fsetArr[0], tsdbTFileSetClear);
×
1011
    taosMemoryFree(fsetArr[0]);
×
1012
    fsetArr[0] = NULL;
×
1013
  }
1014
  return code;
65✔
1015
}
1016

1017
void tsdbFSDestroyCopySnapshot(TFileSetArray **fsetArr) {
65✔
1018
  if (fsetArr[0]) {
65!
1019
    TARRAY2_DESTROY(fsetArr[0], tsdbTFileSetClear);
65!
1020
    taosMemoryFree(fsetArr[0]);
65!
1021
    fsetArr[0] = NULL;
65✔
1022
  }
1023
}
65✔
1024

1025
int32_t tsdbFSCreateRefSnapshot(STFileSystem *fs, TFileSetArray **fsetArr) {
67✔
1026
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
67✔
1027
  int32_t code = tsdbFSCreateRefSnapshotWithoutLock(fs, fsetArr);
67✔
1028
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);
67✔
1029
  return code;
67✔
1030
}
1031

1032
int32_t tsdbFSCreateRefSnapshotWithoutLock(STFileSystem *fs, TFileSetArray **fsetArr) {
944,867✔
1033
  int32_t    code = 0;
944,867✔
1034
  STFileSet *fset, *fset1;
1035

1036
  fsetArr[0] = taosMemoryCalloc(1, sizeof(*fsetArr[0]));
944,867!
1037
  if (fsetArr[0] == NULL) return terrno;
945,220!
1038

1039
  TARRAY2_FOREACH(fs->fSetArr, fset) {
2,029,922✔
1040
    code = tsdbTFileSetInitRef(fs->tsdb, fset, &fset1);
1,084,219✔
1041
    if (code) break;
1,084,711!
1042

1043
    code = TARRAY2_APPEND(fsetArr[0], fset1);
1,084,711✔
1044
    if (code) {
1,084,702!
1045
      tsdbTFileSetClear(&fset1);
×
1046
      break;
×
1047
    }
1048
  }
1049

1050
  if (code) {
945,703!
1051
    TARRAY2_DESTROY(fsetArr[0], tsdbTFileSetClear);
×
1052
    taosMemoryFree(fsetArr[0]);
×
1053
    fsetArr[0] = NULL;
×
1054
  }
1055
  return code;
945,703✔
1056
}
1057

1058
void tsdbFSDestroyRefSnapshot(TFileSetArray **fsetArr) {
944,971✔
1059
  if (fsetArr[0]) {
944,971!
1060
    TARRAY2_DESTROY(fsetArr[0], tsdbTFileSetClear);
2,030,444!
1061
    taosMemoryFreeClear(fsetArr[0]);
945,211!
1062
    fsetArr[0] = NULL;
945,365✔
1063
  }
1064
}
945,245✔
1065

1066
static SHashObj *tsdbFSetRangeArrayToHash(TFileSetRangeArray *pRanges) {
×
1067
  int32_t   capacity = TARRAY2_SIZE(pRanges) * 2;
×
1068
  SHashObj *pHash = taosHashInit(capacity, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
×
1069
  if (pHash == NULL) {
×
1070
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1071
    return NULL;
×
1072
  }
1073

1074
  for (int32_t i = 0; i < TARRAY2_SIZE(pRanges); i++) {
×
1075
    STFileSetRange *u = TARRAY2_GET(pRanges, i);
×
1076
    int32_t         fid = u->fid;
×
1077
    int32_t         code = taosHashPut(pHash, &fid, sizeof(fid), u, sizeof(*u));
×
1078
    tsdbDebug("range diff hash fid:%d, sver:%" PRId64 ", ever:%" PRId64, u->fid, u->sver, u->ever);
×
1079
  }
1080
  return pHash;
×
1081
}
1082

1083
int32_t tsdbFSCreateCopyRangedSnapshot(STFileSystem *fs, TFileSetRangeArray *pRanges, TFileSetArray **fsetArr,
×
1084
                                       TFileOpArray *fopArr) {
1085
  int32_t    code = 0;
×
1086
  STFileSet *fset;
1087
  STFileSet *fset1;
1088
  SHashObj  *pHash = NULL;
×
1089

1090
  fsetArr[0] = taosMemoryMalloc(sizeof(TFileSetArray));
×
1091
  if (fsetArr == NULL) return terrno;
×
1092
  TARRAY2_INIT(fsetArr[0]);
×
1093

1094
  if (pRanges) {
×
1095
    pHash = tsdbFSetRangeArrayToHash(pRanges);
×
1096
    if (pHash == NULL) {
×
1097
      code = TSDB_CODE_OUT_OF_MEMORY;
×
1098
      goto _out;
×
1099
    }
1100
  }
1101

1102
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
×
1103
  TARRAY2_FOREACH(fs->fSetArr, fset) {
×
1104
    int64_t ever = VERSION_MAX;
×
1105
    if (pHash) {
×
1106
      int32_t         fid = fset->fid;
×
1107
      STFileSetRange *u = taosHashGet(pHash, &fid, sizeof(fid));
×
1108
      if (u) {
×
1109
        ever = u->sver - 1;
×
1110
      }
1111
    }
1112

1113
    code = tsdbTFileSetFilteredInitDup(fs->tsdb, fset, ever, &fset1, fopArr);
×
1114
    if (code) break;
×
1115

1116
    code = TARRAY2_APPEND(fsetArr[0], fset1);
×
1117
    if (code) break;
×
1118
  }
1119
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);
×
1120

1121
_out:
×
1122
  if (code) {
×
1123
    TARRAY2_DESTROY(fsetArr[0], tsdbTFileSetClear);
×
1124
    taosMemoryFree(fsetArr[0]);
×
1125
    fsetArr[0] = NULL;
×
1126
  }
1127
  if (pHash) {
×
1128
    taosHashCleanup(pHash);
×
1129
    pHash = NULL;
×
1130
  }
1131
  return code;
×
1132
}
1133

1134
void tsdbFSDestroyCopyRangedSnapshot(TFileSetArray **fsetArr) { tsdbFSDestroyCopySnapshot(fsetArr); }
×
1135

1136
int32_t tsdbFSCreateRefRangedSnapshot(STFileSystem *fs, int64_t sver, int64_t ever, TFileSetRangeArray *pRanges,
×
1137
                                      TFileSetRangeArray **fsrArr) {
1138
  int32_t         code = 0;
×
1139
  STFileSet      *fset;
1140
  STFileSetRange *fsr1 = NULL;
×
1141
  SHashObj       *pHash = NULL;
×
1142

1143
  fsrArr[0] = taosMemoryCalloc(1, sizeof(*fsrArr[0]));
×
1144
  if (fsrArr[0] == NULL) {
×
1145
    code = terrno;
×
1146
    goto _out;
×
1147
  }
1148

1149
  tsdbInfo("pRanges size:%d", (pRanges == NULL ? 0 : TARRAY2_SIZE(pRanges)));
×
1150
  if (pRanges) {
×
1151
    pHash = tsdbFSetRangeArrayToHash(pRanges);
×
1152
    if (pHash == NULL) {
×
1153
      code = TSDB_CODE_OUT_OF_MEMORY;
×
1154
      goto _out;
×
1155
    }
1156
  }
1157

1158
  (void)taosThreadMutexLock(&fs->tsdb->mutex);
×
1159
  TARRAY2_FOREACH(fs->fSetArr, fset) {
×
1160
    int64_t sver1 = sver;
×
1161
    int64_t ever1 = ever;
×
1162

1163
    if (pHash) {
×
1164
      int32_t         fid = fset->fid;
×
1165
      STFileSetRange *u = taosHashGet(pHash, &fid, sizeof(fid));
×
1166
      if (u) {
×
1167
        sver1 = u->sver;
×
1168
        tsdbDebug("range hash get fid:%d, sver:%" PRId64 ", ever:%" PRId64, u->fid, u->sver, u->ever);
×
1169
      }
1170
    }
1171

1172
    if (sver1 > ever1) {
×
1173
      tsdbDebug("skip fid:%d, sver:%" PRId64 ", ever:%" PRId64, fset->fid, sver1, ever1);
×
1174
      continue;
×
1175
    }
1176

1177
    tsdbDebug("fsrArr:%p, fid:%d, sver:%" PRId64 ", ever:%" PRId64, fsrArr, fset->fid, sver1, ever1);
×
1178

1179
    code = tsdbTFileSetRangeInitRef(fs->tsdb, fset, sver1, ever1, &fsr1);
×
1180
    if (code) break;
×
1181

1182
    code = TARRAY2_APPEND(fsrArr[0], fsr1);
×
1183
    if (code) break;
×
1184

1185
    fsr1 = NULL;
×
1186
  }
1187
  (void)taosThreadMutexUnlock(&fs->tsdb->mutex);
×
1188

1189
  if (code) {
×
1190
    tsdbTFileSetRangeClear(&fsr1);
×
1191
    TARRAY2_DESTROY(fsrArr[0], tsdbTFileSetRangeClear);
×
1192
    fsrArr[0] = NULL;
×
1193
  }
1194

1195
_out:
×
1196
  if (pHash) {
×
1197
    taosHashCleanup(pHash);
×
1198
    pHash = NULL;
×
1199
  }
1200
  return code;
×
1201
}
1202

1203
void tsdbFSDestroyRefRangedSnapshot(TFileSetRangeArray **fsrArr) { tsdbTFileSetRangeArrayDestroy(fsrArr); }
×
1204

1205
void tsdbBeginTaskOnFileSet(STsdb *tsdb, int32_t fid, STFileSet **fset) {
129,059✔
1206
  int16_t sttTrigger = tsdb->pVnode->config.sttTrigger;
129,059✔
1207

1208
  tsdbFSGetFSet(tsdb->pFS, fid, fset);
129,059✔
1209
  if (sttTrigger == 1 && (*fset)) {
129,059✔
1210
    for (;;) {
1211
      if ((*fset)->taskRunning) {
292!
1212
        (*fset)->numWaitTask++;
×
1213

1214
        (void)taosThreadCondWait(&(*fset)->beginTask, &tsdb->mutex);
×
1215

1216
        tsdbFSGetFSet(tsdb->pFS, fid, fset);
×
1217

1218
        (*fset)->numWaitTask--;
×
1219
      } else {
1220
        (*fset)->taskRunning = true;
292✔
1221
        break;
292✔
1222
      }
1223
    }
1224
    tsdbInfo("vgId:%d begin task on file set:%d", TD_VID(tsdb->pVnode), fid);
292!
1225
  }
1226
}
129,059✔
1227

1228
void tsdbFinishTaskOnFileSet(STsdb *tsdb, int32_t fid) {
2,320✔
1229
  int16_t sttTrigger = tsdb->pVnode->config.sttTrigger;
2,320✔
1230
  if (sttTrigger == 1) {
2,320✔
1231
    STFileSet *fset = NULL;
292✔
1232
    tsdbFSGetFSet(tsdb->pFS, fid, &fset);
292✔
1233
    if (fset != NULL && fset->taskRunning) {
292!
1234
      fset->taskRunning = false;
292✔
1235
      if (fset->numWaitTask > 0) {
292!
1236
        (void)taosThreadCondSignal(&fset->beginTask);
×
1237
      }
1238
      tsdbInfo("vgId:%d finish task on file set:%d", TD_VID(tsdb->pVnode), fid);
292!
1239
    }
1240
  }
1241
}
2,320✔
1242

1243
struct SFileSetReader {
1244
  STsdb     *pTsdb;
1245
  STFileSet *pFileSet;
1246
  int32_t    fid;
1247
  int64_t    startTime;
1248
  int64_t    endTime;
1249
  int64_t    lastCompactTime;
1250
  int64_t    totalSize;
1251
};
1252

1253
int32_t tsdbFileSetReaderOpen(void *pVnode, struct SFileSetReader **ppReader) {
×
1254
  if (pVnode == NULL || ppReader == NULL) {
×
1255
    return TSDB_CODE_INVALID_PARA;
×
1256
  }
1257

1258
  STsdb *pTsdb = ((SVnode *)pVnode)->pTsdb;
×
1259

1260
  (*ppReader) = taosMemoryCalloc(1, sizeof(struct SFileSetReader));
×
1261
  if (*ppReader == NULL) {
×
1262
    tsdbError("vgId:%d %s failed at %s:%d since %s", TD_VID(pTsdb->pVnode), __func__, __FILE__, __LINE__,
×
1263
              tstrerror(terrno));
1264
    return terrno;
×
1265
  }
1266

1267
  (*ppReader)->pTsdb = pTsdb;
×
1268
  (*ppReader)->fid = INT32_MIN;
×
1269
  (*ppReader)->pFileSet = NULL;
×
1270

1271
  return TSDB_CODE_SUCCESS;
×
1272
}
1273

1274
static int32_t tsdbFileSetReaderNextNoLock(struct SFileSetReader *pReader) {
×
1275
  STsdb  *pTsdb = pReader->pTsdb;
×
1276
  int32_t code = TSDB_CODE_SUCCESS;
×
1277

1278
  tsdbTFileSetClear(&pReader->pFileSet);
×
1279

1280
  STFileSet *fset = &(STFileSet){
×
1281
      .fid = pReader->fid,
×
1282
  };
1283

1284
  STFileSet **fsetPtr = TARRAY2_SEARCH(pReader->pTsdb->pFS->fSetArr, &fset, tsdbTFileSetCmprFn, TD_GT);
×
1285
  if (fsetPtr == NULL) {
×
1286
    pReader->fid = INT32_MAX;
×
1287
    return TSDB_CODE_NOT_FOUND;
×
1288
  }
1289

1290
  // ref file set
1291
  code = tsdbTFileSetInitRef(pReader->pTsdb, *fsetPtr, &pReader->pFileSet);
×
1292
  if (code) return code;
×
1293

1294
  // get file set details
1295
  pReader->fid = pReader->pFileSet->fid;
×
1296
  tsdbFidKeyRange(pReader->fid, pTsdb->keepCfg.days, pTsdb->keepCfg.precision, &pReader->startTime, &pReader->endTime);
×
1297
  pReader->lastCompactTime = 0;  // TODO
×
1298
  pReader->totalSize = 0;
×
1299
  for (int32_t i = 0; i < TSDB_FTYPE_MAX; i++) {
×
1300
    STFileObj *fobj = pReader->pFileSet->farr[i];
×
1301
    if (fobj) {
×
1302
      pReader->totalSize += fobj->f->size;
×
1303
    }
1304
  }
1305
  SSttLvl *lvl;
1306
  TARRAY2_FOREACH(pReader->pFileSet->lvlArr, lvl) {
×
1307
    STFileObj *fobj;
1308
    TARRAY2_FOREACH(lvl->fobjArr, fobj) { pReader->totalSize += fobj->f->size; }
×
1309
  }
1310

1311
  return code;
×
1312
}
1313

1314
int32_t tsdbFileSetReaderNext(struct SFileSetReader *pReader) {
×
1315
  int32_t code = TSDB_CODE_SUCCESS;
×
1316
  (void)taosThreadMutexLock(&pReader->pTsdb->mutex);
×
1317
  code = tsdbFileSetReaderNextNoLock(pReader);
×
1318
  (void)taosThreadMutexUnlock(&pReader->pTsdb->mutex);
×
1319
  return code;
×
1320
}
1321

1322
int32_t tsdbFileSetGetEntryField(struct SFileSetReader *pReader, const char *field, void *value) {
×
1323
  const char *fieldName;
1324

1325
  if (pReader->fid == INT32_MIN || pReader->fid == INT32_MAX) {
×
1326
    return TSDB_CODE_INVALID_PARA;
×
1327
  }
1328

1329
  fieldName = "fileset_id";
×
1330
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1331
    *(int32_t *)value = pReader->fid;
×
1332
    return TSDB_CODE_SUCCESS;
×
1333
  }
1334

1335
  fieldName = "start_time";
×
1336
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1337
    *(int64_t *)value = pReader->startTime;
×
1338
    return TSDB_CODE_SUCCESS;
×
1339
  }
1340

1341
  fieldName = "end_time";
×
1342
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1343
    *(int64_t *)value = pReader->endTime;
×
1344
    return TSDB_CODE_SUCCESS;
×
1345
  }
1346

1347
  fieldName = "total_size";
×
1348
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1349
    *(int64_t *)value = pReader->totalSize;
×
1350
    return TSDB_CODE_SUCCESS;
×
1351
  }
1352

1353
  fieldName = "last_compact_time";
×
1354
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1355
    *(int64_t *)value = pReader->lastCompactTime;
×
1356
    return TSDB_CODE_SUCCESS;
×
1357
  }
1358

1359
  fieldName = "should_compact";
×
1360
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1361
    *(char *)value = 0;  // TODO
×
1362
    return TSDB_CODE_SUCCESS;
×
1363
  }
1364

1365
  fieldName = "details";
×
1366
  if (strncmp(field, fieldName, strlen(fieldName) + 1) == 0) {
×
1367
    // TODO
1368
    return TSDB_CODE_SUCCESS;
×
1369
  }
1370

1371
  return TSDB_CODE_INVALID_PARA;
×
1372
}
1373

1374
void tsdbFileSetReaderClose(struct SFileSetReader **ppReader) {
×
1375
  if (ppReader == NULL || *ppReader == NULL) {
×
1376
    return;
×
1377
  }
1378

1379
  tsdbTFileSetClear(&(*ppReader)->pFileSet);
×
1380
  taosMemoryFree(*ppReader);
×
1381

1382
  *ppReader = NULL;
×
1383
  return;
×
1384
}
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