diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 7636d9b9d0..72418148b0 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -190,6 +190,8 @@ typedef struct SRetention { int8_t keepUnit; } SRetention; +#define RETENTION_VALID(r) (((r)->freq > 0) && ((r)->keep > 0)) + #pragma pack(push, 1) // null-terminated string instead of char array to avoid too many memory consumption in case of more than 1M tableMeta diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 45f697f688..2abed74351 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -41,7 +41,6 @@ extern "C" { // vnode typedef struct SVnode SVnode; typedef struct STsdbCfg STsdbCfg; // todo: remove -typedef struct STsdbKeepCfg STsdbKeepCfg; typedef struct SVnodeCfg SVnodeCfg; extern const SVnodeCfg vnodeCfgDefault; @@ -147,21 +146,7 @@ struct STsdbCfg { SRetention retentions[TSDB_RETENTION_MAX]; }; -struct STsdbKeepCfg { - int8_t precision; // precision always use with below cfg - int32_t days; - int32_t keep0; - int32_t keep1; - int32_t keep2; -}; -typedef enum { - TSDB_TYPE_TSDB = 0, // TSDB - TSDB_TYPE_TSMA = 1, // TSMA - TSDB_TYPE_RSMA_L0 = 2, // RSMA Level 0 - TSDB_TYPE_RSMA_L1 = 3, // RSMA Level 1 - TSDB_TYPE_RSMA_L2 = 4, // RSMA Level 2 -} ETsdbType; struct SVnodeCfg { int32_t vgId; diff --git a/source/dnode/vnode/src/inc/tsdb.h b/source/dnode/vnode/src/inc/tsdb.h index b8cbb2d997..14d776ec35 100644 --- a/source/dnode/vnode/src/inc/tsdb.h +++ b/source/dnode/vnode/src/inc/tsdb.h @@ -184,18 +184,18 @@ struct STsdbFS { SFSStatus *nstatus; // new status }; -#define REPO_ID(r) TD_VID((r)->pVnode) -#define REPO_CFG(r) (&(r)->pVnode->config.tsdbCfg) -#define REPO_KEEP_CFG(r) (&(r)->keepCfg) -#define REPO_LEVEL(r) ((r)->level) -#define REPO_FS(r) ((r)->fs) -#define REPO_META(r) ((r)->pVnode->pMeta) -#define REPO_TFS(r) ((r)->pVnode->pTfs) -#define IS_REPO_LOCKED(r) ((r)->repoLocked) -#define REPO_TSMA_NUM(r) ((r)->smaEnvs.nTSma) -#define REPO_RSMA_NUM(r) ((r)->smaEnvs.nRSma) -#define REPO_TSMA_ENV(r) ((r)->smaEnvs.pTSmaEnv) -#define REPO_RSMA_ENV(r) ((r)->smaEnvs.pRSmaEnv) +#define REPO_ID(r) TD_VID((r)->pVnode) +#define REPO_CFG(r) (&(r)->pVnode->config.tsdbCfg) +#define REPO_KEEP_CFG(r) (&(r)->keepCfg) +#define REPO_LEVEL(r) ((r)->level) +#define REPO_FS(r) ((r)->fs) +#define REPO_META(r) ((r)->pVnode->pMeta) +#define REPO_TFS(r) ((r)->pVnode->pTfs) +#define IS_REPO_LOCKED(r) ((r)->repoLocked) +#define REPO_TSMA_NUM(r) ((r)->smaEnvs.nTSma) +#define REPO_RSMA_NUM(r) ((r)->smaEnvs.nRSma) +#define REPO_TSMA_ENV(r) ((r)->smaEnvs.pTSmaEnv) +#define REPO_RSMA_ENV(r) ((r)->smaEnvs.pRSmaEnv) int tsdbLockRepo(STsdb *pTsdb); int tsdbUnlockRepo(STsdb *pTsdb); diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 3724a98a5e..a7617291b0 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -59,6 +59,7 @@ typedef struct SQWorker SQHandle; #define VNODE_TQ_DIR "tq" #define VNODE_WAL_DIR "wal" #define VNODE_TSMA_DIR "tsma" +#define VNODE_RSMA0_DIR "tsdb" #define VNODE_RSMA1_DIR "rsma1" #define VNODE_RSMA2_DIR "rsma2" @@ -154,6 +155,22 @@ struct SVnodeInfo { SVState state; }; +typedef enum { + TSDB_TYPE_TSDB = 0, // TSDB + TSDB_TYPE_TSMA = 1, // TSMA + TSDB_TYPE_RSMA_L0 = 2, // RSMA Level 0 + TSDB_TYPE_RSMA_L1 = 3, // RSMA Level 1 + TSDB_TYPE_RSMA_L2 = 4, // RSMA Level 2 +} ETsdbType; + +typedef struct { + int8_t precision; // precision always be used with below keep cfgs + int32_t days; + int32_t keep0; + int32_t keep1; + int32_t keep2; +} STsdbKeepCfg; + struct SVnode { char* path; SVnodeCfg config; @@ -176,10 +193,11 @@ struct SVnode { SQHandle* pQuery; }; -#define VND_TSDB(vnd) ((vnd)->pTsdb) -#define VND_RSMA0(vnd) ((vnd)->pTsdb) -#define VND_RSMA1(vnd) ((vnd)->pRSma1) -#define VND_RSMA2(vnd) ((vnd)->pRSma2) +#define VND_TSDB(vnd) ((vnd)->pTsdb) +#define VND_RSMA0(vnd) ((vnd)->pTsdb) +#define VND_RSMA1(vnd) ((vnd)->pRSma1) +#define VND_RSMA2(vnd) ((vnd)->pRSma2) +#define VND_RETENTIONS(vnd) (&(vnd)->config.tsdbCfg.retentions) struct STbUidStore { tb_uid_t suid; diff --git a/source/dnode/vnode/src/tsdb/tsdbOpen.c b/source/dnode/vnode/src/tsdb/tsdbOpen.c index a15bf9f3a4..807ee95b03 100644 --- a/source/dnode/vnode/src/tsdb/tsdbOpen.c +++ b/source/dnode/vnode/src/tsdb/tsdbOpen.c @@ -15,9 +15,30 @@ #include "tsdb.h" +#define TSDB_OPEN_RSMA_IMPL(v, l) \ + do { \ + SRetention *r = VND_RETENTIONS(v)[0]; \ + if (RETENTION_VALID(r)) { \ + return tsdbOpenImpl((v), type, &VND_RSMA##l(v), VNODE_RSMA##l##_DIR, TSDB_RETENTION_L##l); \ + } \ + } while (0) -static int tsdbSetKeepCfg(STsdbCfg *pCfg, STsdbKeepCfg *pKeepCfg, int8_t type); -static int tsdbOpenImpl(SVnode *pVnode, int8_t type, STsdb **ppTsdb, const char *dir, int8_t level); +#define TSDB_SET_KEEP_CFG(l) \ + do { \ + SRetention *r = &pCfg->retentions[l]; \ + pKeepCfg->keep2 = convertTimeFromPrecisionToUnit(r->keep, pCfg->precision, TIME_UNIT_MINUTE); \ + pKeepCfg->keep0 = pKeepCfg->keep2; \ + pKeepCfg->keep1 = pKeepCfg->keep2; \ + pKeepCfg->days = tsdbEvalDays(r, pCfg->precision); \ + } while (0) + +#define RETENTION_DAYS_SPLIT_RATIO 10 +#define RETENTION_DAYS_SPLIT_MIN 1 +#define RETENTION_DAYS_SPLIT_MAX 30 + +static int32_t tsdbSetKeepCfg(STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg, int8_t type); +static int32_t tsdbEvalDays(SRetention *r, int8_t precision); +static int32_t tsdbOpenImpl(SVnode *pVnode, int8_t type, STsdb **ppTsdb, const char *dir, int8_t level); int tsdbOpen(SVnode *pVnode, int8_t type) { switch (type) { @@ -27,11 +48,14 @@ int tsdbOpen(SVnode *pVnode, int8_t type) { ASSERT(0); break; case TSDB_TYPE_RSMA_L0: - return tsdbOpenImpl(pVnode, type, &VND_RSMA0(pVnode), VNODE_TSDB_DIR, TSDB_RETENTION_L0); + TSDB_OPEN_RSMA_IMPL(pVnode, 0); + break; case TSDB_TYPE_RSMA_L1: - return tsdbOpenImpl(pVnode, type, &VND_RSMA1(pVnode), VNODE_RSMA1_DIR, TSDB_RETENTION_L1); + TSDB_OPEN_RSMA_IMPL(pVnode, 1); + break; case TSDB_TYPE_RSMA_L2: - return tsdbOpenImpl(pVnode, type, &VND_RSMA2(pVnode), VNODE_RSMA2_DIR, TSDB_RETENTION_L2); + TSDB_OPEN_RSMA_IMPL(pVnode, 2); + break; default: ASSERT(0); break; @@ -39,7 +63,28 @@ int tsdbOpen(SVnode *pVnode, int8_t type) { return 0; } -static int tsdbSetKeepCfg(STsdbCfg *pCfg, STsdbKeepCfg *pKeepCfg, int8_t type) { +static int32_t tsdbEvalDays(SRetention *r, int8_t precision) { + int32_t keepDays = convertTimeFromPrecisionToUnit(r->keep, precision, TIME_UNIT_DAY); + int32_t freqDays = convertTimeFromPrecisionToUnit(r->freq, precision, TIME_UNIT_DAY); + + int32_t days = keepDays / RETENTION_DAYS_SPLIT_RATIO; + if (days <= RETENTION_DAYS_SPLIT_MIN) { + days = RETENTION_DAYS_SPLIT_MIN; + if (days < freqDays) { + days = freqDays + 1; + } + } else { + if (days > RETENTION_DAYS_SPLIT_MAX) { + days = RETENTION_DAYS_SPLIT_MAX; + } + if (days < freqDays) { + days = freqDays + 1; + } + } + return days * 1440; +} + +static int32_t tsdbSetKeepCfg(STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg, int8_t type) { pKeepCfg->precision = pCfg->precision; switch (type) { case TSDB_TYPE_TSDB: @@ -52,22 +97,13 @@ static int tsdbSetKeepCfg(STsdbCfg *pCfg, STsdbKeepCfg *pKeepCfg, int8_t type) { ASSERT(0); break; case TSDB_TYPE_RSMA_L0: - pKeepCfg->days = pCfg->days; - pKeepCfg->keep0 = pCfg->keep0; - pKeepCfg->keep1 = pCfg->keep1; - pKeepCfg->keep2 = pCfg->keep2; + TSDB_SET_KEEP_CFG(0); break; case TSDB_TYPE_RSMA_L1: - pKeepCfg->days = pCfg->days; - pKeepCfg->keep0 = pCfg->keep0; - pKeepCfg->keep1 = pCfg->keep1; - pKeepCfg->keep2 = pCfg->keep2; + TSDB_SET_KEEP_CFG(1); break; case TSDB_TYPE_RSMA_L2: - pKeepCfg->days = pCfg->days; - pKeepCfg->keep0 = pCfg->keep0; - pKeepCfg->keep1 = pCfg->keep1; - pKeepCfg->keep2 = pCfg->keep2; + TSDB_SET_KEEP_CFG(2); break; default: ASSERT(0); @@ -77,16 +113,16 @@ static int tsdbSetKeepCfg(STsdbCfg *pCfg, STsdbKeepCfg *pKeepCfg, int8_t type) { } /** - * @brief - * - * @param pVnode - * @param type - * @param ppTsdb - * @param dir + * @brief + * + * @param pVnode + * @param type + * @param ppTsdb + * @param dir * @param level retention level - * @return int + * @return int */ -int tsdbOpenImpl(SVnode *pVnode, int8_t type, STsdb **ppTsdb, const char *dir, int8_t level) { +int32_t tsdbOpenImpl(SVnode *pVnode, int8_t type, STsdb **ppTsdb, const char *dir, int8_t level) { STsdb *pTsdb = NULL; int slen = 0; @@ -101,13 +137,12 @@ int tsdbOpenImpl(SVnode *pVnode, int8_t type, STsdb **ppTsdb, const char *dir, i } pTsdb->path = (char *)&pTsdb[1]; - sprintf(pTsdb->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, - dir); + sprintf(pTsdb->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, dir); pTsdb->pVnode = pVnode; pTsdb->level = level; pTsdb->repoLocked = false; taosThreadMutexInit(&pTsdb->mutex, NULL); - tsdbSetKeepCfg(REPO_CFG(pTsdb), REPO_KEEP_CFG(pTsdb), type); + tsdbSetKeepCfg(REPO_KEEP_CFG(pTsdb), REPO_CFG(pTsdb), type); pTsdb->fs = tsdbNewFS(REPO_KEEP_CFG(pTsdb)); // create dir (TODO: use tfsMkdir)