#5049

Committed 11 May 2026 06:30AM UTC coverage: 73.313% (+0.09%) from 73.222%

Build # #5049

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat: refactor taosdump code to improve backup speed and compression ratio (#35292)

Coverage Stats

6625 of 8435 new or added lines in 28 files covered. (78.54%)

2491 existing lines in 142 files now uncovered.

281233 of 383605 relevant lines covered (73.31%)

132489999.79 hits per line

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

87.97

/source/dnode/mnode/sdb/src/sdb.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/>.
 */

#define _DEFAULT_SOURCE
#include "sdb.h"

static int32_t sdbCreateDir(SSdb *pSdb);

SSdb *sdbInit(SSdbOpt *pOption) {
  mInfo("start to init sdb in %s", pOption->path);

  SSdb *pSdb = taosMemoryCalloc(1, sizeof(SSdb));
  if (pSdb == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to init sdb since %s", terrstr());
    return NULL;
  }

  char path[PATH_MAX + 100] = {0};
  snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP);
  pSdb->currDir = taosStrdup(path);
  snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP);
  pSdb->tmpDir = taosStrdup(path);

  // Store mnode directory path for persisting encrypted flag
  tstrncpy(pSdb->mnodePath, pOption->path, PATH_MAX);

  if (pSdb->currDir == NULL || pSdb->tmpDir == NULL) {
    sdbCleanup(pSdb);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to init sdb since %s", terrstr());
    return NULL;
  }

  if (sdbCreateDir(pSdb) != 0) {
    sdbCleanup(pSdb);
    return NULL;
  }

  for (ESdbType i = 0; i < SDB_MAX; ++i) {
    (void)taosThreadRwlockInit(&pSdb->locks[i], NULL);
    pSdb->maxId[i] = 0;
    pSdb->tableVer[i] = 0;
    pSdb->keyTypes[i] = SDB_KEY_INT32;
  }

  pSdb->pWal = pOption->pWal;
  pSdb->applyIndex = -1;
  pSdb->applyTerm = -1;
  pSdb->applyConfig = -1;
  pSdb->commitIndex = -1;
  pSdb->commitTerm = -1;
  pSdb->commitConfig = -1;
  pSdb->pMnode = pOption->pMnode;
  pSdb->encrypted = false;
  (void)taosThreadMutexInit(&pSdb->filelock, NULL);
  mInfo("sdb init success");
  return pSdb;
}

void sdbCleanup(SSdb *pSdb) {
  mInfo("start to cleanup sdb");

  int32_t code = 0;

  if ((code = sdbWriteFile(pSdb, 0)) != 0) {
    mError("failed to write sdb file since %s", tstrerror(code));
  }

  if (pSdb->currDir != NULL) {
    taosMemoryFreeClear(pSdb->currDir);
  }

  if (pSdb->tmpDir != NULL) {
    taosRemoveDir(pSdb->tmpDir);
    taosMemoryFreeClear(pSdb->tmpDir);
  }

  // mnodePath is now a fixed-size array (char mnodePath[PATH_MAX]), no need to free

  for (ESdbType i = 0; i < SDB_MAX; ++i) {
    SHashObj *hash = pSdb->hashObjs[i];
    if (hash == NULL) continue;

    SSdbRow **ppRow = taosHashIterate(hash, NULL);
    while (ppRow != NULL) {
      SSdbRow *pRow = *ppRow;
      if (pRow == NULL) continue;

      sdbFreeRow(pSdb, pRow, true);
      ppRow = taosHashIterate(hash, ppRow);
    }
  }

  for (ESdbType i = 0; i < SDB_MAX; ++i) {
    SHashObj *hash = pSdb->hashObjs[i];
    if (hash == NULL) continue;

    taosHashClear(hash);
    taosHashCleanup(hash);
    (void)taosThreadRwlockDestroy(&pSdb->locks[i]);
    pSdb->hashObjs[i] = NULL;
    memset(&pSdb->locks[i], 0, sizeof(pSdb->locks[i]));

    mInfo("sdb table:%s is cleaned up", sdbTableName(i));
  }

  (void)taosThreadMutexDestroy(&pSdb->filelock);
  taosMemoryFree(pSdb);
  mInfo("sdb is cleaned up");
}

int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {
  int32_t code = 0;

  ESdbType sdbType = table.sdbType;
  EKeyType keyType = table.keyType;
  pSdb->keyTypes[sdbType] = table.keyType;
  pSdb->insertFps[sdbType] = table.insertFp;
  pSdb->updateFps[sdbType] = table.updateFp;
  pSdb->deleteFps[sdbType] = table.deleteFp;
  pSdb->deployFps[sdbType] = table.deployFp;
  pSdb->encodeFps[sdbType] = table.encodeFp;
  pSdb->decodeFps[sdbType] = table.decodeFp;
  pSdb->upgradeFps[sdbType] = table.upgradeFp;
  pSdb->afterRestoredFps[sdbType] = table.afterRestoredFp;
  pSdb->validateFps[sdbType] = table.validateFp;
  pSdb->isUpgradedFps[sdbType] = table.isUpgradedFp;

  int32_t hashType = 0;
  if (keyType == SDB_KEY_INT32) {
    hashType = TSDB_DATA_TYPE_INT;
  } else if (keyType == SDB_KEY_INT64) {
    hashType = TSDB_DATA_TYPE_BIGINT;
  } else {
    hashType = TSDB_DATA_TYPE_BINARY;
  }

  SHashObj *hash = taosHashInit(64, taosGetDefaultHashFunction(hashType), true, HASH_ENTRY_LOCK);
  if (hash == NULL) {
    TAOS_RETURN(terrno);
  }

  pSdb->maxId[sdbType] = 0;
  pSdb->hashObjs[sdbType] = hash;
  mInfo("sdb table:%s is initialized", sdbTableName(sdbType));

  TAOS_RETURN(0);
}

static int32_t sdbCreateDir(SSdb *pSdb) {
  int32_t code = 0;
  if (taosMulMkDir(pSdb->currDir) != 0) {
    code = TAOS_SYSTEM_ERROR(ERRNO);
    mError("failed to create dir:%s since %s", pSdb->currDir, tstrerror(code));
    TAOS_RETURN(code);
  }

  if (taosMkDir(pSdb->tmpDir) != 0) {
    code = TAOS_SYSTEM_ERROR(ERRNO);
    mError("failed to create dir:%s since %s", pSdb->tmpDir, tstrerror(code));
    TAOS_RETURN(code);
  }

  return 0;
}

void sdbSetApplyInfo(SSdb *pSdb, int64_t index, int64_t term, int64_t config) {
  mInfo("vgId:1, mnode apply info changed from index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " to index:%" PRId64
        " term:%" PRId64 " config:%" PRId64,
        pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, index, term, config);

  pSdb->applyIndex = index;
  pSdb->applyTerm = term;
  pSdb->applyConfig = config;
}

void sdbGetCommitInfo(SSdb *pSdb, int64_t *index, int64_t *term, int64_t *config) {
  *index = pSdb->commitIndex;
  *term = pSdb->commitTerm;
  *config = pSdb->commitConfig;
#if 1
  mTrace("mnode current info, apply index:%" PRId64 " term:%" PRId64 " config:%" PRId64 ", commit index:%" PRId64
         " term:%" PRId64 " config:%" PRId64,
         pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, *index, *term, *config);
#endif
}

void sdbWriteLock(SSdb *pSdb, int32_t type) {
  TdThreadRwlock *pLock = &pSdb->locks[type];
  // mTrace("sdb table:%d start write lock:%p", type, pLock);
  (void)taosThreadRwlockWrlock(pLock);
  // mTrace("sdb table:%d stop write lock:%p", type, pLock);
}

void sdbReadLock(SSdb *pSdb, int32_t type) {
  TdThreadRwlock *pLock = &pSdb->locks[type];
  // mTrace("sdb table:%d start read lock:%p", type, pLock);
  (void)taosThreadRwlockRdlock(pLock);
  // mTrace("sdb table:%d stop read lock:%p", type, pLock);
}

void sdbUnLock(SSdb *pSdb, int32_t type) {
  TdThreadRwlock *pLock = &pSdb->locks[type];
  // mTrace("sdb table:%d unlock:%p", type, pLock);
  (void)taosThreadRwlockUnlock(pLock);
}

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	#define _DEFAULT_SOURCE
17	#include "sdb.h"
18
19	static int32_t sdbCreateDir(SSdb *pSdb);
20
21	SSdb sdbInit(SSdbOpt pOption) {	543,612✔
22	mInfo("start to init sdb in %s", pOption->path);	543,612✔
23
24	SSdb *pSdb = taosMemoryCalloc(1, sizeof(SSdb));	543,612✔
25	if (pSdb == NULL) {	543,612✔
26	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
27	mError("failed to init sdb since %s", terrstr());	×
28	return NULL;	×
29	}
30
31	char path[PATH_MAX + 100] = {0};	543,612✔
32	snprintf(path, sizeof(path), "%s%sdata", pOption->path, TD_DIRSEP);	543,612✔
33	pSdb->currDir = taosStrdup(path);	543,612✔
34	snprintf(path, sizeof(path), "%s%stmp", pOption->path, TD_DIRSEP);	543,612✔
35	pSdb->tmpDir = taosStrdup(path);	543,612✔
36
37	// Store mnode directory path for persisting encrypted flag
38	tstrncpy(pSdb->mnodePath, pOption->path, PATH_MAX);	543,612✔
39
40	if (pSdb->currDir == NULL \|\| pSdb->tmpDir == NULL) {	543,612✔
41	sdbCleanup(pSdb);	×
42	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
43	mError("failed to init sdb since %s", terrstr());	×
44	return NULL;	×
45	}
46
47	if (sdbCreateDir(pSdb) != 0) {	543,612✔
48	sdbCleanup(pSdb);	×
49	return NULL;	×
50	}
51
52	for (ESdbType i = 0; i < SDB_MAX; ++i) {	27,180,600✔
53	(void)taosThreadRwlockInit(&pSdb->locks[i], NULL);	26,636,988✔
54	pSdb->maxId[i] = 0;	26,636,988✔
55	pSdb->tableVer[i] = 0;	26,636,988✔
56	pSdb->keyTypes[i] = SDB_KEY_INT32;	26,636,988✔
57	}
58
59	pSdb->pWal = pOption->pWal;	543,612✔
60	pSdb->applyIndex = -1;	543,612✔
61	pSdb->applyTerm = -1;	543,612✔
62	pSdb->applyConfig = -1;	543,612✔
63	pSdb->commitIndex = -1;	543,612✔
64	pSdb->commitTerm = -1;	543,612✔
65	pSdb->commitConfig = -1;	543,612✔
66	pSdb->pMnode = pOption->pMnode;	543,612✔
67	pSdb->encrypted = false;	543,612✔
68	(void)taosThreadMutexInit(&pSdb->filelock, NULL);	543,612✔
69	mInfo("sdb init success");	543,612✔
70	return pSdb;	543,612✔
71	}
72
73	void sdbCleanup(SSdb *pSdb) {	543,347✔
74	mInfo("start to cleanup sdb");	543,347✔
75
76	int32_t code = 0;	543,347✔
77
78	if ((code = sdbWriteFile(pSdb, 0)) != 0) {	543,347✔
79	mError("failed to write sdb file since %s", tstrerror(code));	4,215✔
80	}
81
82	if (pSdb->currDir != NULL) {	543,347✔
83	taosMemoryFreeClear(pSdb->currDir);	543,347✔
84	}
85
86	if (pSdb->tmpDir != NULL) {	543,347✔
87	taosRemoveDir(pSdb->tmpDir);	543,347✔
88	taosMemoryFreeClear(pSdb->tmpDir);	543,347✔
89	}
90
91	// mnodePath is now a fixed-size array (char mnodePath[PATH_MAX]), no need to free
92
93	for (ESdbType i = 0; i < SDB_MAX; ++i) {	27,167,350✔
94	SHashObj *hash = pSdb->hashObjs[i];	26,624,003✔
95	if (hash == NULL) continue;	26,624,003✔
96
97	SSdbRow **ppRow = taosHashIterate(hash, NULL);	24,390,423✔
98	while (ppRow != NULL) {	109,922,685✔
99	SSdbRow pRow = ppRow;	85,532,262✔
100	if (pRow == NULL) continue;	85,532,262✔
101
102	sdbFreeRow(pSdb, pRow, true);	85,532,262✔
103	ppRow = taosHashIterate(hash, ppRow);	85,532,262✔
104	}
105	}
106
107	for (ESdbType i = 0; i < SDB_MAX; ++i) {	27,167,350✔
108	SHashObj *hash = pSdb->hashObjs[i];	26,624,003✔
109	if (hash == NULL) continue;	26,624,003✔
110
111	taosHashClear(hash);	24,390,423✔
112	taosHashCleanup(hash);	24,390,423✔
113	(void)taosThreadRwlockDestroy(&pSdb->locks[i]);	24,390,423✔
114	pSdb->hashObjs[i] = NULL;	24,390,423✔
115	memset(&pSdb->locks[i], 0, sizeof(pSdb->locks[i]));	24,390,423✔
116
117	mInfo("sdb table:%s is cleaned up", sdbTableName(i));	24,390,423✔
118	}
119
120	(void)taosThreadMutexDestroy(&pSdb->filelock);	543,347✔
121	taosMemoryFree(pSdb);	543,347✔
122	mInfo("sdb is cleaned up");	543,347✔
123	}	543,347✔
124
125	int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {	24,393,636✔
126	int32_t code = 0;	24,393,636✔
127
128	ESdbType sdbType = table.sdbType;	24,393,636✔
129	EKeyType keyType = table.keyType;	24,393,636✔
130	pSdb->keyTypes[sdbType] = table.keyType;	24,393,636✔
131	pSdb->insertFps[sdbType] = table.insertFp;	24,393,636✔
132	pSdb->updateFps[sdbType] = table.updateFp;	24,393,636✔
133	pSdb->deleteFps[sdbType] = table.deleteFp;	24,393,636✔
134	pSdb->deployFps[sdbType] = table.deployFp;	24,393,636✔
135	pSdb->encodeFps[sdbType] = table.encodeFp;	24,393,636✔
136	pSdb->decodeFps[sdbType] = table.decodeFp;	24,393,636✔
137	pSdb->upgradeFps[sdbType] = table.upgradeFp;	24,393,636✔
138	pSdb->afterRestoredFps[sdbType] = table.afterRestoredFp;	24,393,636✔
139	pSdb->validateFps[sdbType] = table.validateFp;	24,393,636✔
140	pSdb->isUpgradedFps[sdbType] = table.isUpgradedFp;	24,393,636✔
141
142	int32_t hashType = 0;	24,393,636✔
143	if (keyType == SDB_KEY_INT32) {	24,393,636✔
144	hashType = TSDB_DATA_TYPE_INT;	11,925,012✔
145	} else if (keyType == SDB_KEY_INT64) {	12,468,624✔
146	hashType = TSDB_DATA_TYPE_BIGINT;	2,710,536✔
147	} else {
148	hashType = TSDB_DATA_TYPE_BINARY;	9,758,088✔
149	}
150
151	SHashObj *hash = taosHashInit(64, taosGetDefaultHashFunction(hashType), true, HASH_ENTRY_LOCK);	24,393,636✔
152	if (hash == NULL) {	24,393,636✔
UNCOV 153	TAOS_RETURN(terrno);	×
154	}
155
156	pSdb->maxId[sdbType] = 0;	24,393,636✔
157	pSdb->hashObjs[sdbType] = hash;	24,393,636✔
158	mInfo("sdb table:%s is initialized", sdbTableName(sdbType));	24,393,636✔
159
160	TAOS_RETURN(0);	24,393,636✔
161	}
162
163	static int32_t sdbCreateDir(SSdb *pSdb) {	543,612✔
164	int32_t code = 0;	543,612✔
165	if (taosMulMkDir(pSdb->currDir) != 0) {	543,612✔
166	code = TAOS_SYSTEM_ERROR(ERRNO);	×
167	mError("failed to create dir:%s since %s", pSdb->currDir, tstrerror(code));	×
UNCOV 168	TAOS_RETURN(code);	×
169	}
170
171	if (taosMkDir(pSdb->tmpDir) != 0) {	543,612✔
172	code = TAOS_SYSTEM_ERROR(ERRNO);	×
173	mError("failed to create dir:%s since %s", pSdb->tmpDir, tstrerror(code));	×
UNCOV 174	TAOS_RETURN(code);	×
175	}
176
177	return 0;	543,612✔
178	}
179
180	void sdbSetApplyInfo(SSdb *pSdb, int64_t index, int64_t term, int64_t config) {	51,798,256✔
181	mInfo("vgId:1, mnode apply info changed from index:%" PRId64 " term:%" PRId64 " config:%" PRId64 " to index:%" PRId64	51,798,256✔
182	" term:%" PRId64 " config:%" PRId64,
183	pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, index, term, config);
184
185	pSdb->applyIndex = index;	51,798,256✔
186	pSdb->applyTerm = term;	51,798,256✔
187	pSdb->applyConfig = config;	51,798,256✔
188	}	51,798,256✔
189
190	void sdbGetCommitInfo(SSdb pSdb, int64_t index, int64_t term, int64_t config) {	13,956,569✔
191	*index = pSdb->commitIndex;	13,956,569✔
192	*term = pSdb->commitTerm;	13,956,569✔
193	*config = pSdb->commitConfig;	13,956,569✔
194	#if 1
195	mTrace("mnode current info, apply index:%" PRId64 " term:%" PRId64 " config:%" PRId64 ", commit index:%" PRId64	13,956,569✔
196	" term:%" PRId64 " config:%" PRId64,
197	pSdb->applyIndex, pSdb->applyTerm, pSdb->applyConfig, index, term, *config);
198	#endif
199	}	13,956,569✔
200
201	void sdbWriteLock(SSdb *pSdb, int32_t type) {	2,147,483,647✔
202	TdThreadRwlock *pLock = &pSdb->locks[type];	2,147,483,647✔
203	// mTrace("sdb table:%d start write lock:%p", type, pLock);
204	(void)taosThreadRwlockWrlock(pLock);	2,147,483,647✔
205	// mTrace("sdb table:%d stop write lock:%p", type, pLock);
206	}	2,147,483,647✔
207
208	void sdbReadLock(SSdb *pSdb, int32_t type) {	2,147,483,647✔
209	TdThreadRwlock *pLock = &pSdb->locks[type];	2,147,483,647✔
210	// mTrace("sdb table:%d start read lock:%p", type, pLock);
211	(void)taosThreadRwlockRdlock(pLock);	2,147,483,647✔
212	// mTrace("sdb table:%d stop read lock:%p", type, pLock);
213	}	2,147,483,647✔
214
215	void sdbUnLock(SSdb *pSdb, int32_t type) {	2,147,483,647✔
216	TdThreadRwlock *pLock = &pSdb->locks[type];	2,147,483,647✔
217	// mTrace("sdb table:%d unlock:%p", type, pLock);
218	(void)taosThreadRwlockUnlock(pLock);	2,147,483,647✔
219	}	2,147,483,647✔

taosdata / TDengine / #5049

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