diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 01281a6dc75884fa0672c42b02de876da5cedaf1..92119ecf0de77f6dc5f8e4cb9e1fa6da7ff0e02d 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -119,6 +119,7 @@ extern bool tsQueryUseNodeAllocator; extern bool tsKeepColumnName; extern bool tsEnableQueryHb; extern bool tsEnableScience; +extern bool tsTtlChangeOnWrite; extern int32_t tsRedirectPeriod; extern int32_t tsRedirectFactor; extern int32_t tsRedirectMaxPeriod; diff --git a/include/common/tmsg.h b/include/common/tmsg.h index d78e771fcf08af0f864ce42a3b944ab93b16b833..b8ea35df27229beed9c7162f73ebf25ca97f23b8 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -945,7 +945,7 @@ int32_t tSerializeSVTrimDbReq(void* buf, int32_t bufLen, SVTrimDbReq* pReq); int32_t tDeserializeSVTrimDbReq(void* buf, int32_t bufLen, SVTrimDbReq* pReq); typedef struct { - int32_t timestamp; + int32_t timestampSec; } SVDropTtlTableReq; int32_t tSerializeSVDropTtlTableReq(void* buf, int32_t bufLen, SVDropTtlTableReq* pReq); @@ -2261,7 +2261,7 @@ typedef struct SVCreateTbReq { int32_t flags; char* name; tb_uid_t uid; - int64_t ctime; + int64_t btime; int32_t ttl; int32_t commentLen; char* comment; @@ -2398,10 +2398,12 @@ typedef struct { int32_t newTTL; int32_t newCommentLen; char* newComment; + int64_t ctimeMs; // fill by vnode } SVAlterTbReq; int32_t tEncodeSVAlterTbReq(SEncoder* pEncoder, const SVAlterTbReq* pReq); int32_t tDecodeSVAlterTbReq(SDecoder* pDecoder, SVAlterTbReq* pReq); +int32_t tDecodeSVAlterTbReqSetCtime(SDecoder* pDecoder, SVAlterTbReq* pReq, int64_t ctimeMs); typedef struct { int32_t code; @@ -3405,6 +3407,7 @@ typedef struct SDeleteRes { int64_t affectedRows; char tableFName[TSDB_TABLE_NAME_LEN]; char tsColName[TSDB_COL_NAME_LEN]; + int64_t ctimeMs; // fill by vnode } SDeleteRes; int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes); @@ -3423,10 +3426,12 @@ int32_t tDecodeSSingleDeleteReq(SDecoder* pCoder, SSingleDeleteReq* pReq); typedef struct { int64_t suid; SArray* deleteReqs; // SArray + int64_t ctimeMs; // fill by vnode } SBatchDeleteReq; int32_t tEncodeSBatchDeleteReq(SEncoder* pCoder, const SBatchDeleteReq* pReq); int32_t tDecodeSBatchDeleteReq(SDecoder* pCoder, SBatchDeleteReq* pReq); +int32_t tDecodeSBatchDeleteReqSetCtime(SDecoder* pDecoder, SBatchDeleteReq* pReq, int64_t ctimeMs); typedef struct { int32_t msgIdx; @@ -3496,6 +3501,7 @@ typedef struct { SArray* aRowP; SArray* aCol; }; + int64_t ctimeMs; } SSubmitTbData; typedef struct { diff --git a/include/libs/executor/storageapi.h b/include/libs/executor/storageapi.h index e2944d13dadf12391fd256846ab23cf7fb11e3fc..6031b99cfcef2259123457a792e711b3d80e9004 100644 --- a/include/libs/executor/storageapi.h +++ b/include/libs/executor/storageapi.h @@ -54,7 +54,7 @@ typedef struct SMetaEntry { SRSmaParam rsmaParam; } stbEntry; struct { - int64_t ctime; + int64_t btime; int32_t ttlDays; int32_t commentLen; char* comment; @@ -62,7 +62,7 @@ typedef struct SMetaEntry { uint8_t* pTags; } ctbEntry; struct { - int64_t ctime; + int64_t btime; int32_t ttlDays; int32_t commentLen; char* comment; diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index a79351d5cc73ff423e7b1ab0295008b8a99daaed..716e0f0a3e11ad99408f69b3d9eb9fda07257634 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -108,6 +108,7 @@ int32_t tsQueryRspPolicy = 0; int64_t tsQueryMaxConcurrentTables = 200; // unit is TSDB_TABLE_NUM_UNIT bool tsEnableQueryHb = false; bool tsEnableScience = false; // on taos-cli show float and doulbe with scientific notation if true +bool tsTtlChangeOnWrite = false; // ttl delete time changes on last write if true int32_t tsQuerySmaOptimize = 0; int32_t tsQueryRsmaTolerance = 1000; // the tolerance time (ms) to judge from which level to query rsma data. bool tsQueryPlannerTrace = false; @@ -511,6 +512,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "ttlUnit", tsTtlUnit, 1, 86400 * 365, 1) != 0) return -1; if (cfgAddInt32(pCfg, "ttlPushInterval", tsTtlPushInterval, 1, 100000, 1) != 0) return -1; + if (cfgAddBool(pCfg, "ttlChangeOnWrite", tsTtlChangeOnWrite, 0) != 0) return -1; if (cfgAddInt32(pCfg, "uptimeInterval", tsUptimeInterval, 1, 100000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "queryRsmaTolerance", tsQueryRsmaTolerance, 0, 900000, 0) != 0) return -1; @@ -871,6 +873,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsEnableTelem = cfgGetItem(pCfg, "telemetryReporting")->bval; tsEnableCrashReport = cfgGetItem(pCfg, "crashReporting")->bval; + tsTtlChangeOnWrite = cfgGetItem(pCfg, "ttlChangeOnWrite")->bval; tsTelemInterval = cfgGetItem(pCfg, "telemetryInterval")->i32; tstrncpy(tsTelemServer, cfgGetItem(pCfg, "telemetryServer")->str, TSDB_FQDN_LEN); tsTelemPort = (uint16_t)cfgGetItem(pCfg, "telemetryPort")->i32; @@ -976,6 +979,8 @@ int32_t taosApplyLocalCfg(SConfig *pCfg, char *name) { taosSetCoreDump(enableCore); } else if (strcasecmp("enableQueryHb", name) == 0) { tsEnableQueryHb = cfgGetItem(pCfg, "enableQueryHb")->bval; + } else if (strcasecmp("ttlChangeOnWrite", name) == 0) { + tsTtlChangeOnWrite = cfgGetItem(pCfg, "ttlChangeOnWrite")->bval; } break; } diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index ac035e0a2b9ac631bff120d34c204732df28acde..f55f07835d3f46cbfc7d17dc50cc1784133dffd3 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -30,6 +30,9 @@ #include "tlog.h" +static int32_t tDecodeSVAlterTbReqCommon(SDecoder *pDecoder, SVAlterTbReq *pReq); +static int32_t tDecodeSBatchDeleteReqCommon(SDecoder *pDecoder, SBatchDeleteReq *pReq); + int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) { if (pMsg == NULL) { terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP; @@ -1723,7 +1726,7 @@ int32_t tDeserializeSDropDnodeReq(void *buf, int32_t bufLen, SDropDnodeReq *pReq } else { pReq->unsafe = false; } - + tEndDecode(&decoder); tDecoderClear(&decoder); @@ -3159,7 +3162,7 @@ int32_t tSerializeSVDropTtlTableReq(void *buf, int32_t bufLen, SVDropTtlTableReq tEncoderInit(&encoder, buf, bufLen); if (tStartEncode(&encoder) < 0) return -1; - if (tEncodeI32(&encoder, pReq->timestamp) < 0) return -1; + if (tEncodeI32(&encoder, pReq->timestampSec) < 0) return -1; tEndEncode(&encoder); int32_t tlen = encoder.pos; @@ -3172,7 +3175,7 @@ int32_t tDeserializeSVDropTtlTableReq(void *buf, int32_t bufLen, SVDropTtlTableR tDecoderInit(&decoder, buf, bufLen); if (tStartDecode(&decoder) < 0) return -1; - if (tDecodeI32(&decoder, &pReq->timestamp) < 0) return -1; + if (tDecodeI32(&decoder, &pReq->timestampSec) < 0) return -1; tEndDecode(&decoder); tDecoderClear(&decoder); @@ -4669,7 +4672,7 @@ int32_t tDeserializeSAlterVnodeReplicaReq(void *buf, int32_t bufLen, SAlterVnode if (tDecodeSReplica(&decoder, pReplica) < 0) return -1; } } - + tEndDecode(&decoder); tDecoderClear(&decoder); return 0; @@ -6357,7 +6360,7 @@ int tEncodeSVCreateTbReq(SEncoder *pCoder, const SVCreateTbReq *pReq) { if (tEncodeI32v(pCoder, pReq->flags) < 0) return -1; if (tEncodeCStr(pCoder, pReq->name) < 0) return -1; if (tEncodeI64(pCoder, pReq->uid) < 0) return -1; - if (tEncodeI64(pCoder, pReq->ctime) < 0) return -1; + if (tEncodeI64(pCoder, pReq->btime) < 0) return -1; if (tEncodeI32(pCoder, pReq->ttl) < 0) return -1; if (tEncodeI8(pCoder, pReq->type) < 0) return -1; if (tEncodeI32(pCoder, pReq->commentLen) < 0) return -1; @@ -6392,7 +6395,7 @@ int tDecodeSVCreateTbReq(SDecoder *pCoder, SVCreateTbReq *pReq) { if (tDecodeI32v(pCoder, &pReq->flags) < 0) return -1; if (tDecodeCStr(pCoder, &pReq->name) < 0) return -1; if (tDecodeI64(pCoder, &pReq->uid) < 0) return -1; - if (tDecodeI64(pCoder, &pReq->ctime) < 0) return -1; + if (tDecodeI64(pCoder, &pReq->btime) < 0) return -1; if (tDecodeI32(pCoder, &pReq->ttl) < 0) return -1; if (tDecodeI8(pCoder, &pReq->type) < 0) return -1; if (tDecodeI32(pCoder, &pReq->commentLen) < 0) return -1; @@ -6857,14 +6860,13 @@ int32_t tEncodeSVAlterTbReq(SEncoder *pEncoder, const SVAlterTbReq *pReq) { default: break; } + if (tEncodeI64(pEncoder, pReq->ctimeMs) < 0) return -1; tEndEncode(pEncoder); return 0; } -int32_t tDecodeSVAlterTbReq(SDecoder *pDecoder, SVAlterTbReq *pReq) { - if (tStartDecode(pDecoder) < 0) return -1; - +static int32_t tDecodeSVAlterTbReqCommon(SDecoder *pDecoder, SVAlterTbReq *pReq) { if (tDecodeCStr(pDecoder, &pReq->tbName) < 0) return -1; if (tDecodeI8(pDecoder, &pReq->action) < 0) return -1; if (tDecodeI32(pDecoder, &pReq->colId) < 0) return -1; @@ -6908,6 +6910,28 @@ int32_t tDecodeSVAlterTbReq(SDecoder *pDecoder, SVAlterTbReq *pReq) { default: break; } + return 0; +} + +int32_t tDecodeSVAlterTbReq(SDecoder *pDecoder, SVAlterTbReq *pReq) { + if (tStartDecode(pDecoder) < 0) return -1; + if (tDecodeSVAlterTbReqCommon(pDecoder, pReq) < 0) return -1; + + pReq->ctimeMs = 0; + if (!tDecodeIsEnd(pDecoder)) { + if (tDecodeI64(pDecoder, &pReq->ctimeMs) < 0) return -1; + } + + tEndDecode(pDecoder); + return 0; +} + +int32_t tDecodeSVAlterTbReqSetCtime(SDecoder* pDecoder, SVAlterTbReq* pReq, int64_t ctimeMs) { + if (tStartDecode(pDecoder) < 0) return -1; + if (tDecodeSVAlterTbReqCommon(pDecoder, pReq) < 0) return -1; + + *(int64_t *)(pDecoder->data + pDecoder->pos) = ctimeMs; + if (tDecodeI64(pDecoder, &pReq->ctimeMs) < 0) return -1; tEndDecode(pDecoder); return 0; @@ -7186,6 +7210,7 @@ int32_t tEncodeDeleteRes(SEncoder *pCoder, const SDeleteRes *pRes) { if (tEncodeCStr(pCoder, pRes->tableFName) < 0) return -1; if (tEncodeCStr(pCoder, pRes->tsColName) < 0) return -1; + if (tEncodeI64(pCoder, pRes->ctimeMs) < 0) return -1; return 0; } @@ -7205,6 +7230,11 @@ int32_t tDecodeDeleteRes(SDecoder *pCoder, SDeleteRes *pRes) { if (tDecodeCStrTo(pCoder, pRes->tableFName) < 0) return -1; if (tDecodeCStrTo(pCoder, pRes->tsColName) < 0) return -1; + + pRes->ctimeMs = 0; + if (!tDecodeIsEnd(pCoder)) { + if (tDecodeI64(pCoder, &pRes->ctimeMs) < 0) return -1; + } return 0; } @@ -7428,10 +7458,11 @@ int32_t tEncodeSBatchDeleteReq(SEncoder *pEncoder, const SBatchDeleteReq *pReq) SSingleDeleteReq *pOneReq = taosArrayGet(pReq->deleteReqs, i); if (tEncodeSSingleDeleteReq(pEncoder, pOneReq) < 0) return -1; } + if (tEncodeI64(pEncoder, pReq->ctimeMs) < 0) return -1; return 0; } -int32_t tDecodeSBatchDeleteReq(SDecoder *pDecoder, SBatchDeleteReq *pReq) { +static int32_t tDecodeSBatchDeleteReqCommon(SDecoder *pDecoder, SBatchDeleteReq *pReq) { if (tDecodeI64(pDecoder, &pReq->suid) < 0) return -1; int32_t sz; if (tDecodeI32(pDecoder, &sz) < 0) return -1; @@ -7445,6 +7476,24 @@ int32_t tDecodeSBatchDeleteReq(SDecoder *pDecoder, SBatchDeleteReq *pReq) { return 0; } +int32_t tDecodeSBatchDeleteReq(SDecoder *pDecoder, SBatchDeleteReq *pReq) { + if (tDecodeSBatchDeleteReqCommon(pDecoder, pReq)) return -1; + + pReq->ctimeMs = 0; + if (!tDecodeIsEnd(pDecoder)) { + if (tDecodeI64(pDecoder, &pReq->ctimeMs) < 0) return -1; + } + return 0; +} + +int32_t tDecodeSBatchDeleteReqSetCtime(SDecoder *pDecoder, SBatchDeleteReq *pReq, int64_t ctimeMs) { + if (tDecodeSBatchDeleteReqCommon(pDecoder, pReq)) return -1; + + *(int64_t *)(pDecoder->data + pDecoder->pos) = ctimeMs; + if (tDecodeI64(pDecoder, &pReq->ctimeMs) < 0) return -1; + return 0; +} + static int32_t tEncodeSSubmitTbData(SEncoder *pCoder, const SSubmitTbData *pSubmitTbData) { if (tStartEncode(pCoder) < 0) return -1; @@ -7479,6 +7528,7 @@ static int32_t tEncodeSSubmitTbData(SEncoder *pCoder, const SSubmitTbData *pSubm pCoder->pos += rows[iRow]->len; } } + if (tEncodeI64(pCoder, pSubmitTbData->ctimeMs) < 0) return -1; tEndEncode(pCoder); return 0; @@ -7559,6 +7609,14 @@ static int32_t tDecodeSSubmitTbData(SDecoder *pCoder, SSubmitTbData *pSubmitTbDa } } + pSubmitTbData->ctimeMs = 0; + if (!tDecodeIsEnd(pCoder)) { + if (tDecodeI64(pCoder, &pSubmitTbData->ctimeMs) < 0) { + code = TSDB_CODE_INVALID_MSG; + goto _exit; + } + } + tEndDecode(pCoder); _exit: diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index d58b9fd4bf16ba9f4a632f4266a453b34c9e8866..b843ad61a9ead8e9c5ad4fcc2818f579ada0f6f1 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -888,7 +888,7 @@ static int32_t mndProcessTtlTimer(SRpcMsg *pReq) { SSdb *pSdb = pMnode->pSdb; SVgObj *pVgroup = NULL; void *pIter = NULL; - SVDropTtlTableReq ttlReq = {.timestamp = taosGetTimestampSec()}; + SVDropTtlTableReq ttlReq = {.timestampSec = taosGetTimestampSec()}; int32_t reqLen = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq); int32_t contLen = reqLen + sizeof(SMsgHead); @@ -914,7 +914,7 @@ static int32_t mndProcessTtlTimer(SRpcMsg *pReq) { if (code != 0) { mError("vgId:%d, failed to send drop ttl table request to vnode since 0x%x", pVgroup->vgId, code); } else { - mInfo("vgId:%d, send drop ttl table request to vnode, time:%d", pVgroup->vgId, ttlReq.timestamp); + mInfo("vgId:%d, send drop ttl table request to vnode, time:%" PRId32, pVgroup->vgId, ttlReq.timestampSec); } sdbRelease(pSdb, pVgroup); } @@ -1188,7 +1188,7 @@ static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *p if (mndAllocStbSchemas(pOld, pNew) != 0) { return -1; } - + if(pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ntags){ terrno = TSDB_CODE_MND_FIELD_VALUE_OVERFLOW; return -1; diff --git a/source/dnode/vnode/CMakeLists.txt b/source/dnode/vnode/CMakeLists.txt index b7bfc57cd5e720a918c5a5613c5cd8d03be4a028..fd4e68ed855655ddf15fd275ad206858a1740434 100644 --- a/source/dnode/vnode/CMakeLists.txt +++ b/source/dnode/vnode/CMakeLists.txt @@ -27,6 +27,7 @@ target_sources( "src/meta/metaEntry.c" "src/meta/metaSnapshot.c" "src/meta/metaCache.c" + "src/meta/metaTtl.c" # sma "src/sma/smaEnv.c" diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 7e19425d56492018577aa733920906874931ebfe..9de6fea76b001ca513755df54d632c53789b0682 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -114,6 +114,7 @@ int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName); int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName); int metaGetTableUidByName(void *pVnode, char *tbName, uint64_t *uid); int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType); +int metaGetTableTtlByUid(void *meta, uint64_t uid, int64_t *ttlDays); bool metaIsTableExist(void* pVnode, tb_uid_t uid); int32_t metaGetCachedTableUidList(void *pVnode, tb_uid_t suid, const uint8_t *key, int32_t keyLen, SArray *pList, bool *acquired); diff --git a/source/dnode/vnode/src/inc/meta.h b/source/dnode/vnode/src/inc/meta.h index b39008147b2d1284a9e6d7562c2463874854ce94..92012b34245db122e68f6fca5d94a6b9e7fc16a8 100644 --- a/source/dnode/vnode/src/inc/meta.h +++ b/source/dnode/vnode/src/inc/meta.h @@ -17,6 +17,7 @@ #define _TD_VNODE_META_H_ #include "index.h" +#include "ttl.h" #include "vnodeInt.h" #ifdef __cplusplus @@ -89,10 +90,10 @@ struct SMeta { // ivt idx and idx void* pTagIvtIdx; - TTB* pTagIdx; - TTB* pTtlIdx; + TTB* pTagIdx; + STtlManger* pTtlMgr; - TTB* pCtimeIdx; // table created time idx + TTB* pBtimeIdx; // table created time idx TTB* pNcolIdx; // ncol of table idx, normal table only TTB* pSmaIdx; @@ -138,20 +139,15 @@ typedef struct { } STagIdxKey; #pragma pack(pop) -typedef struct { - int64_t dtime; - tb_uid_t uid; -} STtlIdxKey; - typedef struct { tb_uid_t uid; int64_t smaUid; } SSmaIdxKey; typedef struct { - int64_t ctime; + int64_t btime; tb_uid_t uid; -} SCtimeIdxKey; +} SBtimeIdxKey; typedef struct { int64_t ncol; diff --git a/source/dnode/vnode/src/inc/metaTtl.h b/source/dnode/vnode/src/inc/metaTtl.h new file mode 100644 index 0000000000000000000000000000000000000000..bf3b897c6f7771b0877a08a440b442f3ee247f84 --- /dev/null +++ b/source/dnode/vnode/src/inc/metaTtl.h @@ -0,0 +1,98 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + +#ifndef _TD_VNODE_TTL_H_ +#define _TD_VNODE_TTL_H_ + +#include "taosdef.h" +#include "thash.h" + +#include "tdb.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef enum DirtyEntryType { + ENTRY_TYPE_DEL = 1, + ENTRY_TYPE_UPSERT = 2, +} DirtyEntryType; + +typedef struct STtlManger { + TdThreadRwlock lock; + + TTB* pOldTtlIdx; // btree<{deleteTime, tuid}, NULL> + + SHashObj* pTtlCache; // key: tuid, value: {ttl, ctime} + SHashObj* pDirtyUids; // dirty tuid + TTB* pTtlIdx; // btree<{deleteTime, tuid}, ttl> +} STtlManger; + +typedef struct { + int64_t ttlDays; + int64_t changeTimeMs; +} STtlCacheEntry; + +typedef struct { + DirtyEntryType type; +} STtlDirtyEntry; + +typedef struct { + int64_t deleteTimeSec; + tb_uid_t uid; +} STtlIdxKey; + +typedef struct { + int64_t deleteTimeMs; + tb_uid_t uid; +} STtlIdxKeyV1; + +typedef struct { + int64_t ttlDays; +} STtlIdxValue; + +typedef struct { + tb_uid_t uid; + int64_t changeTimeMs; +} STtlUpdCtimeCtx; + +typedef struct { + tb_uid_t uid; + int64_t changeTimeMs; + int64_t ttlDays; +} STtlUpdTtlCtx; + +typedef struct { + tb_uid_t uid; + TXN* pTxn; +} STtlDelTtlCtx; + +int ttlMgrOpen(STtlManger** ppTtlMgr, TDB* pEnv, int8_t rollback); +int ttlMgrClose(STtlManger* pTtlMgr); +int ttlMgrBegin(STtlManger* pTtlMgr, void* pMeta); + +int ttlMgrConvert(TTB* pOldTtlIdx, TTB* pNewTtlIdx, void* pMeta); +int ttlMgrFlush(STtlManger* pTtlMgr, TXN* pTxn); + +int ttlMgrInsertTtl(STtlManger* pTtlMgr, const STtlUpdTtlCtx* pUpdCtx); +int ttlMgrDeleteTtl(STtlManger* pTtlMgr, const STtlDelTtlCtx* pDelCtx); +int ttlMgrUpdateChangeTime(STtlManger* pTtlMgr, const STtlUpdCtimeCtx* pUpdCtimeCtx); +int ttlMgrFindExpired(STtlManger* pTtlMgr, int64_t timePointMs, SArray* pTbUids); + +#ifdef __cplusplus +} +#endif + +#endif /*_TD_VNODE_TTL_H_*/ diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 2811fc35b0af49a423679953a2df42b2d1087e0e..d4b2887a1143da0a98864676e387da1c84aa3f4a 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -149,8 +149,9 @@ int metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq int metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq, STableMetaRsp** pMetaRsp); int metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq, SArray* tbUids, int64_t* tbUid); int32_t metaTrimTables(SMeta* pMeta); -int metaTtlDropTable(SMeta* pMeta, int64_t ttl, SArray* tbUids); +int metaTtlDropTable(SMeta* pMeta, int64_t timePointMs, SArray* tbUids); int metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq, STableMetaRsp* pMetaRsp); +int metaUpdateChangeTime(SMeta* pMeta, tb_uid_t uid, int64_t changeTimeMs); SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, int lock); STSchema* metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, int lock); int32_t metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema); @@ -176,7 +177,6 @@ SArray* metaGetSmaIdsByTable(SMeta* pMeta, tb_uid_t uid); SArray* metaGetSmaTbUids(SMeta* pMeta); void* metaGetIdx(SMeta* pMeta); void* metaGetIvtIdx(SMeta* pMeta); -int metaTtlSmaller(SMeta* pMeta, uint64_t time, SArray* uidList); void metaReaderInit(SMetaReader* pReader, SMeta* pMeta, int32_t flags); diff --git a/source/dnode/vnode/src/meta/metaCommit.c b/source/dnode/vnode/src/meta/metaCommit.c index f597c100d0588dae9123af444dc442c0b7de2c2b..1fa5b9c1e9e98434c3c0769c5005c1a11902442a 100644 --- a/source/dnode/vnode/src/meta/metaCommit.c +++ b/source/dnode/vnode/src/meta/metaCommit.c @@ -40,6 +40,12 @@ int metaBegin(SMeta *pMeta, int8_t heap) { return -1; } + if (ttlMgrBegin(pMeta->pTtlMgr, pMeta) < 0) { + return -1; + } + + tdbCommit(pMeta->pEnv, pMeta->txn); + return 0; } @@ -50,6 +56,7 @@ int metaFinishCommit(SMeta *pMeta, TXN *txn) { return tdbPostCommit(pMeta->pEnv int metaPrepareAsyncCommit(SMeta *pMeta) { // return tdbPrepareAsyncCommit(pMeta->pEnv, pMeta->txn); int code = 0; + code = ttlMgrFlush(pMeta->pTtlMgr, pMeta->txn); code = tdbCommit(pMeta->pEnv, pMeta->txn); return code; diff --git a/source/dnode/vnode/src/meta/metaEntry.c b/source/dnode/vnode/src/meta/metaEntry.c index e50931ac0618252eea58afc0d1a734d252a05d1e..01877a523a942c9a3e5bcf889046358334ef9621 100644 --- a/source/dnode/vnode/src/meta/metaEntry.c +++ b/source/dnode/vnode/src/meta/metaEntry.c @@ -31,7 +31,7 @@ int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) { if (tEncodeSRSmaParam(pCoder, &pME->stbEntry.rsmaParam) < 0) return -1; } } else if (pME->type == TSDB_CHILD_TABLE) { - if (tEncodeI64(pCoder, pME->ctbEntry.ctime) < 0) return -1; + if (tEncodeI64(pCoder, pME->ctbEntry.btime) < 0) return -1; if (tEncodeI32(pCoder, pME->ctbEntry.ttlDays) < 0) return -1; if (tEncodeI32v(pCoder, pME->ctbEntry.commentLen) < 0) return -1; if (pME->ctbEntry.commentLen > 0) { @@ -40,7 +40,7 @@ int metaEncodeEntry(SEncoder *pCoder, const SMetaEntry *pME) { if (tEncodeI64(pCoder, pME->ctbEntry.suid) < 0) return -1; if (tEncodeTag(pCoder, (const STag *)pME->ctbEntry.pTags) < 0) return -1; } else if (pME->type == TSDB_NORMAL_TABLE) { - if (tEncodeI64(pCoder, pME->ntbEntry.ctime) < 0) return -1; + if (tEncodeI64(pCoder, pME->ntbEntry.btime) < 0) return -1; if (tEncodeI32(pCoder, pME->ntbEntry.ttlDays) < 0) return -1; if (tEncodeI32v(pCoder, pME->ntbEntry.commentLen) < 0) return -1; if (pME->ntbEntry.commentLen > 0) { @@ -76,7 +76,7 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) { if (tDecodeSRSmaParam(pCoder, &pME->stbEntry.rsmaParam) < 0) return -1; } } else if (pME->type == TSDB_CHILD_TABLE) { - if (tDecodeI64(pCoder, &pME->ctbEntry.ctime) < 0) return -1; + if (tDecodeI64(pCoder, &pME->ctbEntry.btime) < 0) return -1; if (tDecodeI32(pCoder, &pME->ctbEntry.ttlDays) < 0) return -1; if (tDecodeI32v(pCoder, &pME->ctbEntry.commentLen) < 0) return -1; if (pME->ctbEntry.commentLen > 0) { @@ -85,7 +85,7 @@ int metaDecodeEntry(SDecoder *pCoder, SMetaEntry *pME) { if (tDecodeI64(pCoder, &pME->ctbEntry.suid) < 0) return -1; if (tDecodeTag(pCoder, (STag **)&pME->ctbEntry.pTags) < 0) return -1; // (TODO) } else if (pME->type == TSDB_NORMAL_TABLE) { - if (tDecodeI64(pCoder, &pME->ntbEntry.ctime) < 0) return -1; + if (tDecodeI64(pCoder, &pME->ntbEntry.btime) < 0) return -1; if (tDecodeI32(pCoder, &pME->ntbEntry.ttlDays) < 0) return -1; if (tDecodeI32v(pCoder, &pME->ntbEntry.commentLen) < 0) return -1; if (pME->ntbEntry.commentLen > 0) { diff --git a/source/dnode/vnode/src/meta/metaOpen.c b/source/dnode/vnode/src/meta/metaOpen.c index 1d0b11e26a58d047216196bf3fed7b8f13807922..fb17aff31889568a45ed41d8929f0072884accf7 100644 --- a/source/dnode/vnode/src/meta/metaOpen.c +++ b/source/dnode/vnode/src/meta/metaOpen.c @@ -19,12 +19,11 @@ static int tbDbKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen static int skmDbKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int ctbIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int tagIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); -static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int uidIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int smaIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int taskIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); -static int ctimeIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); +static int btimeIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int ncolIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); static int32_t metaInitLock(SMeta *pMeta) { return taosThreadRwlockInit(&pMeta->lock, NULL); } @@ -128,8 +127,8 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) { goto _err; } - // open pTtlIdx - ret = tdbTbOpen("ttl.idx", sizeof(STtlIdxKey), 0, ttlIdxKeyCmpr, pMeta->pEnv, &pMeta->pTtlIdx, 0); + // open pTtlMgr ("ttlv1.idx") + ret = ttlMgrOpen(&pMeta->pTtlMgr, pMeta->pEnv, 0); if (ret < 0) { metaError("vgId:%d, failed to open meta ttl index since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; @@ -143,7 +142,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta, int8_t rollback) { } // idx table create time - ret = tdbTbOpen("ctime.idx", sizeof(SCtimeIdxKey), 0, ctimeIdxCmpr, pMeta->pEnv, &pMeta->pCtimeIdx, 0); + ret = tdbTbOpen("ctime.idx", sizeof(SBtimeIdxKey), 0, btimeIdxCmpr, pMeta->pEnv, &pMeta->pBtimeIdx, 0); if (ret < 0) { metaError("vgId:%d, failed to open meta ctime index since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; @@ -184,9 +183,9 @@ _err: if (pMeta->pIdx) metaCloseIdx(pMeta); if (pMeta->pStreamDb) tdbTbClose(pMeta->pStreamDb); if (pMeta->pNcolIdx) tdbTbClose(pMeta->pNcolIdx); - if (pMeta->pCtimeIdx) tdbTbClose(pMeta->pCtimeIdx); + if (pMeta->pBtimeIdx) tdbTbClose(pMeta->pBtimeIdx); if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx); - if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx); + if (pMeta->pTtlMgr) ttlMgrClose(pMeta->pTtlMgr); if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx); if (pMeta->pTagIdx) tdbTbClose(pMeta->pTagIdx); if (pMeta->pCtbIdx) tdbTbClose(pMeta->pCtbIdx); @@ -209,9 +208,9 @@ int metaClose(SMeta **ppMeta) { if (pMeta->pIdx) metaCloseIdx(pMeta); if (pMeta->pStreamDb) tdbTbClose(pMeta->pStreamDb); if (pMeta->pNcolIdx) tdbTbClose(pMeta->pNcolIdx); - if (pMeta->pCtimeIdx) tdbTbClose(pMeta->pCtimeIdx); + if (pMeta->pBtimeIdx) tdbTbClose(pMeta->pBtimeIdx); if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx); - if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx); + if (pMeta->pTtlMgr) ttlMgrClose(pMeta->pTtlMgr); if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx); if (pMeta->pTagIdx) tdbTbClose(pMeta->pTagIdx); if (pMeta->pCtbIdx) tdbTbClose(pMeta->pCtbIdx); @@ -399,37 +398,18 @@ static int tagIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kL return 0; } -static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) { - STtlIdxKey *pTtlIdxKey1 = (STtlIdxKey *)pKey1; - STtlIdxKey *pTtlIdxKey2 = (STtlIdxKey *)pKey2; - - if (pTtlIdxKey1->dtime > pTtlIdxKey2->dtime) { - return 1; - } else if (pTtlIdxKey1->dtime < pTtlIdxKey2->dtime) { - return -1; - } - - if (pTtlIdxKey1->uid > pTtlIdxKey2->uid) { - return 1; - } else if (pTtlIdxKey1->uid < pTtlIdxKey2->uid) { - return -1; - } - - return 0; -} - -static int ctimeIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) { - SCtimeIdxKey *pCtimeIdxKey1 = (SCtimeIdxKey *)pKey1; - SCtimeIdxKey *pCtimeIdxKey2 = (SCtimeIdxKey *)pKey2; - if (pCtimeIdxKey1->ctime > pCtimeIdxKey2->ctime) { +static int btimeIdxCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) { + SBtimeIdxKey *pBtimeIdxKey1 = (SBtimeIdxKey *)pKey1; + SBtimeIdxKey *pBtimeIdxKey2 = (SBtimeIdxKey *)pKey2; + if (pBtimeIdxKey1->btime > pBtimeIdxKey2->btime) { return 1; - } else if (pCtimeIdxKey1->ctime < pCtimeIdxKey2->ctime) { + } else if (pBtimeIdxKey1->btime < pBtimeIdxKey2->btime) { return -1; } - if (pCtimeIdxKey1->uid > pCtimeIdxKey2->uid) { + if (pBtimeIdxKey1->uid > pBtimeIdxKey2->uid) { return 1; - } else if (pCtimeIdxKey1->uid < pCtimeIdxKey2->uid) { + } else if (pBtimeIdxKey1->uid < pBtimeIdxKey2->uid) { return -1; } diff --git a/source/dnode/vnode/src/meta/metaQuery.c b/source/dnode/vnode/src/meta/metaQuery.c index 29fe89c3f29716bf49940b6bffe061e798a608b1..0e380ea0b2f42614d873a7c41cd32645da35ce84 100644 --- a/source/dnode/vnode/src/meta/metaQuery.c +++ b/source/dnode/vnode/src/meta/metaQuery.c @@ -212,6 +212,29 @@ int metaReadNext(SMetaReader *pReader) { return 0; } +int metaGetTableTtlByUid(void *meta, uint64_t uid, int64_t *ttlDays) { + int code = -1; + SMetaReader mr = {0}; + metaReaderInit(&mr, (SMeta *)meta, 0); + code = metaReaderGetTableEntryByUid(&mr, uid); + if (code < 0) { + goto _exit; + } + if (mr.me.type == TSDB_CHILD_TABLE) { + *ttlDays = mr.me.ctbEntry.ttlDays; + } else if (mr.me.type == TSDB_NORMAL_TABLE) { + *ttlDays = mr.me.ntbEntry.ttlDays; + } else { + goto _exit; + } + + code = 0; + +_exit: + metaReaderClear(&mr); + return code; +} + #if 1 // =================================================== SMTbCursor *metaOpenTbCursor(void *pVnode) { SMTbCursor *pTbCur = NULL; @@ -387,37 +410,6 @@ _err: return NULL; } -int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList) { - TBC *pCur; - int ret = tdbTbcOpen(pMeta->pTtlIdx, &pCur, NULL); - if (ret < 0) { - return ret; - } - - STtlIdxKey ttlKey = {0}; - ttlKey.dtime = ttl; - ttlKey.uid = INT64_MAX; - int c = 0; - tdbTbcMoveTo(pCur, &ttlKey, sizeof(ttlKey), &c); - if (c < 0) { - tdbTbcMoveToPrev(pCur); - } - - void *pKey = NULL; - int kLen = 0; - while (1) { - ret = tdbTbcPrev(pCur, &pKey, &kLen, NULL, NULL); - if (ret < 0) { - break; - } - ttlKey = *(STtlIdxKey *)pKey; - taosArrayPush(uidList, &ttlKey.uid); - } - tdbFree(pKey); - tdbTbcClose(pCur); - return 0; -} - struct SMCtbCursor { SMeta *pMeta; TBC *pCur; @@ -1018,17 +1010,17 @@ int32_t metaFilterCreateTime(void *pVnode, SMetaFltParam *arg, SArray *pUids) { pCursor->type = param->type; metaRLock(pMeta); - ret = tdbTbcOpen(pMeta->pCtimeIdx, &pCursor->pCur, NULL); + ret = tdbTbcOpen(pMeta->pBtimeIdx, &pCursor->pCur, NULL); if (ret != 0) { goto END; } int64_t uidLimit = param->reverse ? INT64_MAX : 0; - SCtimeIdxKey ctimeKey = {.ctime = *(int64_t *)(param->val), .uid = uidLimit}; - SCtimeIdxKey *pCtimeKey = &ctimeKey; + SBtimeIdxKey btimeKey = {.btime = *(int64_t *)(param->val), .uid = uidLimit}; + SBtimeIdxKey *pBtimeKey = &btimeKey; int cmp = 0; - if (tdbTbcMoveTo(pCursor->pCur, &ctimeKey, sizeof(ctimeKey), &cmp) < 0) { + if (tdbTbcMoveTo(pCursor->pCur, &btimeKey, sizeof(btimeKey), &cmp) < 0) { goto END; } @@ -1042,10 +1034,10 @@ int32_t metaFilterCreateTime(void *pVnode, SMetaFltParam *arg, SArray *pUids) { valid = tdbTbcGet(pCursor->pCur, (const void **)&entryKey, &nEntryKey, NULL, NULL); if (valid < 0) break; - SCtimeIdxKey *p = entryKey; + SBtimeIdxKey *p = entryKey; if (count > TRY_ERROR_LIMIT) break; - int32_t cmp = (*param->filterFunc)((void *)&p->ctime, (void *)&pCtimeKey->ctime, param->type); + int32_t cmp = (*param->filterFunc)((void *)&p->btime, (void *)&pBtimeKey->btime, param->type); if (cmp == 0) taosArrayPush(pUids, &p->uid); else { @@ -1149,7 +1141,7 @@ int32_t metaFilterTtl(void *pVnode, SMetaFltParam *arg, SArray *pUids) { pCursor->type = param->type; metaRLock(pMeta); - ret = tdbTbcOpen(pMeta->pTtlIdx, &pCursor->pCur, NULL); + //ret = tdbTbcOpen(pMeta->pTtlIdx, &pCursor->pCur, NULL); END: if (pCursor->pMeta) metaULock(pCursor->pMeta); diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index eb169fbdc20dcbf0ba42040cbec257ac47eb9d12..e3c5d17f93c60615dc12817d3fbf3b22b453602d 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -20,7 +20,7 @@ static int metaDelJsonVarFromIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, con static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateUidIdx(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateNameIdx(SMeta *pMeta, const SMetaEntry *pME); -static int metaUpdateTtlIdx(SMeta *pMeta, const SMetaEntry *pME); +static int metaUpdateTtl(SMeta *pMeta, const SMetaEntry *pME); static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateSuidIdx(SMeta *pMeta, const SMetaEntry *pME); @@ -28,8 +28,8 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry); static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type); static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey); // opt ins_tables query -static int metaUpdateCtimeIdx(SMeta *pMeta, const SMetaEntry *pME); -static int metaDeleteCtimeIdx(SMeta *pMeta, const SMetaEntry *pME); +static int metaUpdateBtimeIdx(SMeta *pMeta, const SMetaEntry *pME); +static int metaDeleteBtimeIdx(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateNcolIdx(SMeta *pMeta, const SMetaEntry *pME); static int metaDeleteNcolIdx(SMeta *pMeta, const SMetaEntry *pME); @@ -734,7 +734,7 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs me.uid = pReq->uid; me.name = pReq->name; if (me.type == TSDB_CHILD_TABLE) { - me.ctbEntry.ctime = pReq->ctime; + me.ctbEntry.btime = pReq->btime; me.ctbEntry.ttlDays = pReq->ttl; me.ctbEntry.commentLen = pReq->commentLen; me.ctbEntry.comment = pReq->comment; @@ -770,7 +770,7 @@ int metaCreateTable(SMeta *pMeta, int64_t ver, SVCreateTbReq *pReq, STableMetaRs metaTbGroupCacheClear(pMeta, me.ctbEntry.suid); metaULock(pMeta); } else { - me.ntbEntry.ctime = pReq->ctime; + me.ntbEntry.btime = pReq->btime; me.ntbEntry.ttlDays = pReq->ttl; me.ntbEntry.commentLen = pReq->commentLen; me.ntbEntry.comment = pReq->comment; @@ -918,50 +918,40 @@ end: return code; } -int metaTtlDropTable(SMeta *pMeta, int64_t ttl, SArray *tbUids) { - int ret = metaTtlSmaller(pMeta, ttl, tbUids); +int metaTtlDropTable(SMeta *pMeta, int64_t timePointMs, SArray *tbUids) { + int ret = ttlMgrFlush(pMeta->pTtlMgr, pMeta->txn); if (ret != 0) { + metaError("ttl failed to flush, ret:%d", ret); + return ret; + } + + ret = ttlMgrFindExpired(pMeta->pTtlMgr, timePointMs, tbUids); + if (ret != 0) { + metaError("ttl failed to find expired table, ret:%d", ret); return ret; } if (TARRAY_SIZE(tbUids) == 0) { return 0; } + metaInfo("ttl find expired table count: %zu" , TARRAY_SIZE(tbUids)); + metaDropTables(pMeta, tbUids); return 0; } -static void metaBuildTtlIdxKey(STtlIdxKey *ttlKey, const SMetaEntry *pME) { - int64_t ttlDays = 0; - int64_t ctime = 0; +static int metaBuildBtimeIdxKey(SBtimeIdxKey *btimeKey, const SMetaEntry *pME) { + int64_t btime; if (pME->type == TSDB_CHILD_TABLE) { - ctime = pME->ctbEntry.ctime; - ttlDays = pME->ctbEntry.ttlDays; + btime = pME->ctbEntry.btime; } else if (pME->type == TSDB_NORMAL_TABLE) { - ctime = pME->ntbEntry.ctime; - ttlDays = pME->ntbEntry.ttlDays; - } else { - metaError("meta/table: invalide table type: %" PRId8 " build ttl idx key failed.", pME->type); - return; - } - - if (ttlDays <= 0) return; - - ttlKey->dtime = ctime / 1000 + ttlDays * tsTtlUnit; - ttlKey->uid = pME->uid; -} -static int metaBuildCtimeIdxKey(SCtimeIdxKey *ctimeKey, const SMetaEntry *pME) { - int64_t ctime; - if (pME->type == TSDB_CHILD_TABLE) { - ctime = pME->ctbEntry.ctime; - } else if (pME->type == TSDB_NORMAL_TABLE) { - ctime = pME->ntbEntry.ctime; + btime = pME->ntbEntry.btime; } else { return -1; } - ctimeKey->ctime = ctime; - ctimeKey->uid = pME->uid; + btimeKey->btime = btime; + btimeKey->uid = pME->uid; return 0; } @@ -975,11 +965,9 @@ static int metaBuildNColIdxKey(SNcolIdxKey *ncolKey, const SMetaEntry *pME) { return 0; } -static int metaDeleteTtlIdx(SMeta *pMeta, const SMetaEntry *pME) { - STtlIdxKey ttlKey = {0}; - metaBuildTtlIdxKey(&ttlKey, pME); - if (ttlKey.dtime == 0) return 0; - return tdbTbDelete(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), pMeta->txn); +static int metaDeleteTtl(SMeta *pMeta, const SMetaEntry *pME) { + STtlDelTtlCtx ctx = {.uid = pME->uid, .pTxn = pMeta->txn}; + return ttlMgrDeleteTtl(pMeta->pTtlMgr, &ctx); } static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { @@ -1061,10 +1049,10 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { tdbTbDelete(pMeta->pNameIdx, e.name, strlen(e.name) + 1, pMeta->txn); tdbTbDelete(pMeta->pUidIdx, &uid, sizeof(uid), pMeta->txn); - if (e.type == TSDB_CHILD_TABLE || e.type == TSDB_NORMAL_TABLE) metaDeleteCtimeIdx(pMeta, &e); + if (e.type == TSDB_CHILD_TABLE || e.type == TSDB_NORMAL_TABLE) metaDeleteBtimeIdx(pMeta, &e); if (e.type == TSDB_NORMAL_TABLE) metaDeleteNcolIdx(pMeta, &e); - if (e.type != TSDB_SUPER_TABLE) metaDeleteTtlIdx(pMeta, &e); + if (e.type != TSDB_SUPER_TABLE) metaDeleteTtl(pMeta, &e); if (e.type == TSDB_CHILD_TABLE) { tdbTbDelete(pMeta->pCtbIdx, &(SCtbIdxKey){.suid = e.ctbEntry.suid, .uid = uid}, sizeof(SCtbIdxKey), pMeta->txn); @@ -1097,23 +1085,23 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { return 0; } // opt ins_tables -int metaUpdateCtimeIdx(SMeta *pMeta, const SMetaEntry *pME) { - SCtimeIdxKey ctimeKey = {0}; - if (metaBuildCtimeIdxKey(&ctimeKey, pME) < 0) { +int metaUpdateBtimeIdx(SMeta *pMeta, const SMetaEntry *pME) { + SBtimeIdxKey btimeKey = {0}; + if (metaBuildBtimeIdxKey(&btimeKey, pME) < 0) { return 0; } - metaTrace("vgId:%d, start to save version:%" PRId64 " uid:%" PRId64 " ctime:%" PRId64, TD_VID(pMeta->pVnode), - pME->version, pME->uid, ctimeKey.ctime); + metaTrace("vgId:%d, start to save version:%" PRId64 " uid:%" PRId64 " btime:%" PRId64, TD_VID(pMeta->pVnode), + pME->version, pME->uid, btimeKey.btime); - return tdbTbUpsert(pMeta->pCtimeIdx, &ctimeKey, sizeof(ctimeKey), NULL, 0, pMeta->txn); + return tdbTbUpsert(pMeta->pBtimeIdx, &btimeKey, sizeof(btimeKey), NULL, 0, pMeta->txn); } -int metaDeleteCtimeIdx(SMeta *pMeta, const SMetaEntry *pME) { - SCtimeIdxKey ctimeKey = {0}; - if (metaBuildCtimeIdxKey(&ctimeKey, pME) < 0) { +int metaDeleteBtimeIdx(SMeta *pMeta, const SMetaEntry *pME) { + SBtimeIdxKey btimeKey = {0}; + if (metaBuildBtimeIdxKey(&btimeKey, pME) < 0) { return 0; } - return tdbTbDelete(pMeta->pCtimeIdx, &ctimeKey, sizeof(ctimeKey), pMeta->txn); + return tdbTbDelete(pMeta->pBtimeIdx, &btimeKey, sizeof(btimeKey), pMeta->txn); } int metaUpdateNcolIdx(SMeta *pMeta, const SMetaEntry *pME) { SNcolIdxKey ncolKey = {0}; @@ -1323,6 +1311,8 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl metaULock(pMeta); + metaUpdateChangeTime(pMeta, entry.uid, pAlterTbReq->ctimeMs); + metaUpdateMetaRsp(uid, pAlterTbReq->tbName, pSchema, pMetaRsp); if (entry.pBuf) taosMemoryFree(entry.pBuf); @@ -1510,6 +1500,8 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA metaULock(pMeta); + metaUpdateChangeTime(pMeta, ctbEntry.uid, pAlterTbReq->ctimeMs); + tDecoderClear(&dc1); tDecoderClear(&dc2); taosMemoryFree((void *)ctbEntry.ctbEntry.pTags); @@ -1598,9 +1590,9 @@ static int metaUpdateTableOptions(SMeta *pMeta, int64_t version, SVAlterTbReq *p // build SMetaEntry if (entry.type == TSDB_CHILD_TABLE) { if (pAlterTbReq->updateTTL) { - metaDeleteTtlIdx(pMeta, &entry); + metaDeleteTtl(pMeta, &entry); entry.ctbEntry.ttlDays = pAlterTbReq->newTTL; - metaUpdateTtlIdx(pMeta, &entry); + metaUpdateTtl(pMeta, &entry); } if (pAlterTbReq->newCommentLen >= 0) { entry.ctbEntry.commentLen = pAlterTbReq->newCommentLen; @@ -1608,9 +1600,9 @@ static int metaUpdateTableOptions(SMeta *pMeta, int64_t version, SVAlterTbReq *p } } else { if (pAlterTbReq->updateTTL) { - metaDeleteTtlIdx(pMeta, &entry); + metaDeleteTtl(pMeta, &entry); entry.ntbEntry.ttlDays = pAlterTbReq->newTTL; - metaUpdateTtlIdx(pMeta, &entry); + metaUpdateTtl(pMeta, &entry); } if (pAlterTbReq->newCommentLen >= 0) { entry.ntbEntry.commentLen = pAlterTbReq->newCommentLen; @@ -1623,6 +1615,8 @@ static int metaUpdateTableOptions(SMeta *pMeta, int64_t version, SVAlterTbReq *p metaUpdateUidIdx(pMeta, &entry); metaULock(pMeta); + metaUpdateChangeTime(pMeta, entry.uid, pAlterTbReq->ctimeMs); + tdbTbcClose(pTbDbc); tdbTbcClose(pUidIdxc); tDecoderClear(&dc); @@ -1962,11 +1956,28 @@ static int metaUpdateNameIdx(SMeta *pMeta, const SMetaEntry *pME) { return tdbTbUpsert(pMeta->pNameIdx, pME->name, strlen(pME->name) + 1, &pME->uid, sizeof(tb_uid_t), pMeta->txn); } -static int metaUpdateTtlIdx(SMeta *pMeta, const SMetaEntry *pME) { - STtlIdxKey ttlKey = {0}; - metaBuildTtlIdxKey(&ttlKey, pME); - if (ttlKey.dtime == 0) return 0; - return tdbTbUpsert(pMeta->pTtlIdx, &ttlKey, sizeof(ttlKey), NULL, 0, pMeta->txn); +static int metaUpdateTtl(SMeta *pMeta, const SMetaEntry *pME) { + if (pME->type != TSDB_CHILD_TABLE && pME->type != TSDB_NORMAL_TABLE) return 0; + + STtlUpdTtlCtx ctx = {.uid = pME->uid}; + + if (pME->type == TSDB_CHILD_TABLE) { + ctx.ttlDays = pME->ctbEntry.ttlDays; + ctx.changeTimeMs = pME->ctbEntry.btime; + } else { + ctx.ttlDays = pME->ntbEntry.ttlDays; + ctx.changeTimeMs = pME->ntbEntry.btime; + } + + return ttlMgrInsertTtl(pMeta->pTtlMgr, &ctx); +} + +int metaUpdateChangeTime(SMeta *pMeta, tb_uid_t uid, int64_t changeTimeMs) { + if (!tsTtlChangeOnWrite) return 0; + + STtlUpdCtimeCtx ctx = {.uid = uid, .changeTimeMs = changeTimeMs}; + + return ttlMgrUpdateChangeTime(pMeta->pTtlMgr, &ctx); } static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME) { @@ -2177,7 +2188,7 @@ int metaHandleEntry(SMeta *pMeta, const SMetaEntry *pME) { } } - code = metaUpdateCtimeIdx(pMeta, pME); + code = metaUpdateBtimeIdx(pMeta, pME); VND_CHECK_CODE(code, line, _err); if (pME->type == TSDB_NORMAL_TABLE) { @@ -2186,7 +2197,7 @@ int metaHandleEntry(SMeta *pMeta, const SMetaEntry *pME) { } if (pME->type != TSDB_SUPER_TABLE) { - code = metaUpdateTtlIdx(pMeta, pME); + code = metaUpdateTtl(pMeta, pME); VND_CHECK_CODE(code, line, _err); } diff --git a/source/dnode/vnode/src/meta/metaTtl.c b/source/dnode/vnode/src/meta/metaTtl.c new file mode 100644 index 0000000000000000000000000000000000000000..a152acf681c268968937dab3daacac93b9bcda12 --- /dev/null +++ b/source/dnode/vnode/src/meta/metaTtl.c @@ -0,0 +1,434 @@ +/* + * Copyright (c) 2019 TAOS Data, Inc. + * + * 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 . + */ + +#include "ttl.h" +#include "meta.h" + +typedef struct { + TTB *pNewTtlIdx; + SMeta *pMeta; +} SConvertData; + +static void ttlMgrBuildKey(STtlIdxKeyV1 *pTtlKey, int64_t ttlDays, int64_t changeTimeMs, tb_uid_t uid); +static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); +static int ttlIdxKeyV1Cmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2); +static int ttlMgrFillCache(STtlManger *pTtlMgr); +static int32_t ttlMgrFillCacheOneEntry(const void *pKey, int keyLen, const void *pVal, int valLen, void *pTtlCache); +static int32_t ttlMgrConvertOneEntry(const void *pKey, int keyLen, const void *pVal, int valLen, void *pConvertData); + +static int32_t ttlMgrWLock(STtlManger *pTtlMgr); +static int32_t ttlMgrRLock(STtlManger *pTtlMgr); +static int32_t ttlMgrULock(STtlManger *pTtlMgr); + +const char *ttlTbname = "ttl.idx"; +const char *ttlV1Tbname = "ttlv1.idx"; + +int ttlMgrOpen(STtlManger **ppTtlMgr, TDB *pEnv, int8_t rollback) { + int ret; + + *ppTtlMgr = NULL; + + STtlManger *pTtlMgr = (STtlManger *)tdbOsCalloc(1, sizeof(*pTtlMgr)); + if (pTtlMgr == NULL) { + return -1; + } + + if (tdbTbExist(ttlTbname, pEnv)) { + ret = tdbTbOpen(ttlTbname, sizeof(STtlIdxKey), 0, ttlIdxKeyCmpr, pEnv, &pTtlMgr->pOldTtlIdx, rollback); + if (ret < 0) { + metaError("failed to open %s index since %s", ttlTbname, tstrerror(terrno)); + return ret; + } + } + + ret = tdbTbOpen(ttlV1Tbname, TDB_VARIANT_LEN, TDB_VARIANT_LEN, ttlIdxKeyV1Cmpr, pEnv, &pTtlMgr->pTtlIdx, rollback); + if (ret < 0) { + metaError("failed to open %s since %s", ttlV1Tbname, tstrerror(terrno)); + + tdbOsFree(pTtlMgr); + return ret; + } + + pTtlMgr->pTtlCache = taosHashInit(8192, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK); + pTtlMgr->pDirtyUids = taosHashInit(8192, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK); + + taosThreadRwlockInit(&pTtlMgr->lock, NULL); + + *ppTtlMgr = pTtlMgr; + return 0; +} + +int ttlMgrClose(STtlManger *pTtlMgr) { + taosHashCleanup(pTtlMgr->pTtlCache); + taosHashCleanup(pTtlMgr->pDirtyUids); + tdbTbClose(pTtlMgr->pTtlIdx); + taosThreadRwlockDestroy(&pTtlMgr->lock); + tdbOsFree(pTtlMgr); + return 0; +} + +int ttlMgrBegin(STtlManger *pTtlMgr, void *pMeta) { + metaInfo("ttl mgr start open"); + int ret; + + int64_t startNs = taosGetTimestampNs(); + + SMeta *meta = (SMeta *)pMeta; + + if (pTtlMgr->pOldTtlIdx) { + ret = ttlMgrConvert(pTtlMgr->pOldTtlIdx, pTtlMgr->pTtlIdx, pMeta); + if (ret < 0) { + metaError("failed to convert ttl index since %s", tstrerror(terrno)); + goto _out; + } + + ret = tdbTbDropByName(ttlTbname, meta->pEnv, meta->txn); + if (ret < 0) { + metaError("failed to drop old ttl index since %s", tstrerror(terrno)); + goto _out; + } + + tdbTbClose(pTtlMgr->pOldTtlIdx); + pTtlMgr->pOldTtlIdx = NULL; + } + + ret = ttlMgrFillCache(pTtlMgr); + if (ret < 0) { + metaError("failed to fill hash since %s", tstrerror(terrno)); + goto _out; + } + + int64_t endNs = taosGetTimestampNs(); + + metaInfo("ttl mgr open end, hash size: %d, time consumed: %" PRId64 " ns", taosHashGetSize(pTtlMgr->pTtlCache), + endNs - startNs); +_out: + return ret; +} + +static void ttlMgrBuildKey(STtlIdxKeyV1 *pTtlKey, int64_t ttlDays, int64_t changeTimeMs, tb_uid_t uid) { + if (ttlDays <= 0) return; + + pTtlKey->deleteTimeMs = changeTimeMs + ttlDays * tsTtlUnit * 1000; + pTtlKey->uid = uid; +} + +static int ttlIdxKeyCmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) { + STtlIdxKey *pTtlIdxKey1 = (STtlIdxKey *)pKey1; + STtlIdxKey *pTtlIdxKey2 = (STtlIdxKey *)pKey2; + + if (pTtlIdxKey1->deleteTimeSec > pTtlIdxKey2->deleteTimeSec) { + return 1; + } else if (pTtlIdxKey1->deleteTimeSec < pTtlIdxKey2->deleteTimeSec) { + return -1; + } + + if (pTtlIdxKey1->uid > pTtlIdxKey2->uid) { + return 1; + } else if (pTtlIdxKey1->uid < pTtlIdxKey2->uid) { + return -1; + } + + return 0; +} + +static int ttlIdxKeyV1Cmpr(const void *pKey1, int kLen1, const void *pKey2, int kLen2) { + STtlIdxKeyV1 *pTtlIdxKey1 = (STtlIdxKeyV1 *)pKey1; + STtlIdxKeyV1 *pTtlIdxKey2 = (STtlIdxKeyV1 *)pKey2; + + if (pTtlIdxKey1->deleteTimeMs > pTtlIdxKey2->deleteTimeMs) { + return 1; + } else if (pTtlIdxKey1->deleteTimeMs < pTtlIdxKey2->deleteTimeMs) { + return -1; + } + + if (pTtlIdxKey1->uid > pTtlIdxKey2->uid) { + return 1; + } else if (pTtlIdxKey1->uid < pTtlIdxKey2->uid) { + return -1; + } + + return 0; +} + +static int ttlMgrFillCache(STtlManger *pTtlMgr) { + return tdbTbTraversal(pTtlMgr->pTtlIdx, pTtlMgr->pTtlCache, ttlMgrFillCacheOneEntry); +} + +static int32_t ttlMgrFillCacheOneEntry(const void *pKey, int keyLen, const void *pVal, int valLen, void *pTtlCache) { + SHashObj *pCache = (SHashObj *)pTtlCache; + + STtlIdxKeyV1 *ttlKey = (STtlIdxKeyV1 *)pKey; + tb_uid_t uid = ttlKey->uid; + int64_t ttlDays = *(int64_t *)pVal; + int64_t changeTimeMs = ttlKey->deleteTimeMs - ttlDays * tsTtlUnit * 1000; + + STtlCacheEntry data = {.ttlDays = ttlDays, .changeTimeMs = changeTimeMs}; + + return taosHashPut(pCache, &uid, sizeof(uid), &data, sizeof(data)); +} + +static int ttlMgrConvertOneEntry(const void *pKey, int keyLen, const void *pVal, int valLen, void *pConvertData) { + SConvertData *pData = (SConvertData *)pConvertData; + + STtlIdxKey *ttlKey = (STtlIdxKey *)pKey; + tb_uid_t uid = ttlKey->uid; + int64_t ttlDays = 0; + + int ret = metaGetTableTtlByUid(pData->pMeta, uid, &ttlDays); + if (ret < 0) { + metaError("ttlMgr convert failed to get ttl since %s", tstrerror(terrno)); + goto _out; + } + + STtlIdxKeyV1 ttlKeyV1 = {.deleteTimeMs = ttlKey->deleteTimeSec * 1000, .uid = uid}; + ret = tdbTbUpsert(pData->pNewTtlIdx, &ttlKeyV1, sizeof(ttlKeyV1), &ttlDays, sizeof(ttlDays), pData->pMeta->txn); + if (ret < 0) { + metaError("ttlMgr convert failed to upsert since %s", tstrerror(terrno)); + goto _out; + } + + ret = 0; +_out: + return ret; +} + +int ttlMgrConvert(TTB *pOldTtlIdx, TTB *pNewTtlIdx, void *pMeta) { + SMeta *meta = pMeta; + + metaInfo("ttlMgr convert ttl start."); + + SConvertData cvData = {.pNewTtlIdx = pNewTtlIdx, .pMeta = meta}; + + int ret = tdbTbTraversal(pOldTtlIdx, &cvData, ttlMgrConvertOneEntry); + if (ret < 0) { + metaError("failed to convert ttl since %s", tstrerror(terrno)); + } + + metaInfo("ttlMgr convert ttl end."); + return ret; +} + +int ttlMgrInsertTtl(STtlManger *pTtlMgr, const STtlUpdTtlCtx *updCtx) { + if (updCtx->ttlDays == 0) return 0; + + STtlCacheEntry cacheEntry = {.ttlDays = updCtx->ttlDays, .changeTimeMs = updCtx->changeTimeMs}; + STtlDirtyEntry dirtryEntry = {.type = ENTRY_TYPE_UPSERT}; + + ttlMgrWLock(pTtlMgr); + + int ret = taosHashPut(pTtlMgr->pTtlCache, &updCtx->uid, sizeof(updCtx->uid), &cacheEntry, sizeof(cacheEntry)); + if (ret < 0) { + metaError("ttlMgr insert failed to update ttl cache since %s", tstrerror(terrno)); + goto _out; + } + + ret = taosHashPut(pTtlMgr->pDirtyUids, &updCtx->uid, sizeof(updCtx->uid), &dirtryEntry, sizeof(dirtryEntry)); + if (ret < 0) { + metaError("ttlMgr insert failed to update ttl dirty uids since %s", tstrerror(terrno)); + goto _out; + } + + ret = 0; +_out: + ttlMgrULock(pTtlMgr); + + metaDebug("ttl mgr insert ttl, uid: %" PRId64 ", ctime: %" PRId64 ", ttlDays: %" PRId64, updCtx->uid, + updCtx->changeTimeMs, updCtx->ttlDays); + + return ret; +} + +int ttlMgrDeleteTtl(STtlManger *pTtlMgr, const STtlDelTtlCtx *delCtx) { + ttlMgrWLock(pTtlMgr); + + STtlDirtyEntry dirtryEntry = {.type = ENTRY_TYPE_DEL}; + + int ret = taosHashPut(pTtlMgr->pDirtyUids, &delCtx->uid, sizeof(delCtx->uid), &dirtryEntry, sizeof(dirtryEntry)); + if (ret < 0) { + metaError("ttlMgr del failed to update ttl dirty uids since %s", tstrerror(terrno)); + goto _out; + } + + ret = 0; +_out: + ttlMgrULock(pTtlMgr); + + metaDebug("ttl mgr delete ttl, uid: %" PRId64, delCtx->uid); + + return ret; +} + +int ttlMgrUpdateChangeTime(STtlManger *pTtlMgr, const STtlUpdCtimeCtx *pUpdCtimeCtx) { + ttlMgrWLock(pTtlMgr); + + STtlCacheEntry *oldData = taosHashGet(pTtlMgr->pTtlCache, &pUpdCtimeCtx->uid, sizeof(pUpdCtimeCtx->uid)); + if (oldData == NULL) { + goto _out; + } + + STtlCacheEntry cacheEntry = {.ttlDays = oldData->ttlDays, .changeTimeMs = pUpdCtimeCtx->changeTimeMs}; + STtlDirtyEntry dirtryEntry = {.type = ENTRY_TYPE_UPSERT}; + + int ret = + taosHashPut(pTtlMgr->pTtlCache, &pUpdCtimeCtx->uid, sizeof(pUpdCtimeCtx->uid), &cacheEntry, sizeof(cacheEntry)); + if (ret < 0) { + metaError("ttlMgr update ctime failed to update ttl cache since %s", tstrerror(terrno)); + goto _out; + } + + ret = taosHashPut(pTtlMgr->pDirtyUids, &pUpdCtimeCtx->uid, sizeof(pUpdCtimeCtx->uid), &dirtryEntry, + sizeof(dirtryEntry)); + if (ret < 0) { + metaError("ttlMgr update ctime failed to update ttl dirty uids since %s", tstrerror(terrno)); + goto _out; + } + + ret = 0; +_out: + ttlMgrULock(pTtlMgr); + + metaDebug("ttl mgr update ctime, uid: %" PRId64 ", ctime: %" PRId64, pUpdCtimeCtx->uid, pUpdCtimeCtx->changeTimeMs); + + return ret; +} + +int ttlMgrFindExpired(STtlManger *pTtlMgr, int64_t timePointMs, SArray *pTbUids) { + ttlMgrRLock(pTtlMgr); + + TBC *pCur; + int ret = tdbTbcOpen(pTtlMgr->pTtlIdx, &pCur, NULL); + if (ret < 0) { + goto _out; + } + + STtlIdxKeyV1 ttlKey = {0}; + ttlKey.deleteTimeMs = timePointMs; + ttlKey.uid = INT64_MAX; + int c = 0; + tdbTbcMoveTo(pCur, &ttlKey, sizeof(ttlKey), &c); + if (c < 0) { + tdbTbcMoveToPrev(pCur); + } + + void *pKey = NULL; + int kLen = 0; + while (1) { + ret = tdbTbcPrev(pCur, &pKey, &kLen, NULL, NULL); + if (ret < 0) { + ret = 0; + break; + } + ttlKey = *(STtlIdxKeyV1 *)pKey; + taosArrayPush(pTbUids, &ttlKey.uid); + } + + tdbFree(pKey); + tdbTbcClose(pCur); + + ret = 0; +_out: + ttlMgrULock(pTtlMgr); + return ret; +} + +int ttlMgrFlush(STtlManger *pTtlMgr, TXN *pTxn) { + ttlMgrWLock(pTtlMgr); + + metaInfo("ttl mgr flush start."); + + int ret = -1; + + void *pIter = taosHashIterate(pTtlMgr->pDirtyUids, NULL); + while (pIter != NULL) { + STtlDirtyEntry *pEntry = (STtlDirtyEntry *)pIter; + tb_uid_t *pUid = taosHashGetKey(pIter, NULL); + + STtlCacheEntry *cacheEntry = taosHashGet(pTtlMgr->pTtlCache, pUid, sizeof(*pUid)); + if (cacheEntry == NULL) { + metaError("ttlMgr flush failed to get ttl cache since %s", tstrerror(terrno)); + goto _out; + } + + STtlIdxKeyV1 ttlKey; + ttlMgrBuildKey(&ttlKey, cacheEntry->ttlDays, cacheEntry->changeTimeMs, *pUid); + + if (pEntry->type == ENTRY_TYPE_UPSERT) { + ret = tdbTbUpsert(pTtlMgr->pTtlIdx, &ttlKey, sizeof(ttlKey), &cacheEntry->ttlDays, sizeof(cacheEntry->ttlDays), + pTxn); + if (ret < 0) { + metaError("ttlMgr flush failed to flush ttl cache upsert since %s", tstrerror(terrno)); + goto _out; + } + } else if (pEntry->type == ENTRY_TYPE_DEL) { + ret = tdbTbDelete(pTtlMgr->pTtlIdx, &ttlKey, sizeof(ttlKey), pTxn); + if (ret < 0) { + metaError("ttlMgr flush failed to flush ttl cache del since %s", tstrerror(terrno)); + goto _out; + } + + ret = taosHashRemove(pTtlMgr->pTtlCache, pUid, sizeof(*pUid)); + if (ret < 0) { + metaError("ttlMgr flush failed to delete ttl cache since %s", tstrerror(terrno)); + goto _out; + } + } else { + metaError("ttlMgr flush failed to flush ttl cache, unknown type: %d", pEntry->type); + goto _out; + } + + pIter = taosHashIterate(pTtlMgr->pDirtyUids, pIter); + } + + taosHashClear(pTtlMgr->pDirtyUids); + + ret = 0; +_out: + ttlMgrULock(pTtlMgr); + + metaInfo("ttl mgr flush end."); + + return ret; +} + +static int32_t ttlMgrRLock(STtlManger *pTtlMgr) { + int32_t ret = 0; + + metaTrace("ttlMgr rlock %p", &pTtlMgr->lock); + + ret = taosThreadRwlockRdlock(&pTtlMgr->lock); + + return ret; +} + +static int32_t ttlMgrWLock(STtlManger *pTtlMgr) { + int32_t ret = 0; + + metaTrace("ttlMgr wlock %p", &pTtlMgr->lock); + + ret = taosThreadRwlockWrlock(&pTtlMgr->lock); + + return ret; +} + +static int32_t ttlMgrULock(STtlManger *pTtlMgr) { + int32_t ret = 0; + + metaTrace("ttlMgr ulock %p", &pTtlMgr->lock); + + ret = taosThreadRwlockUnlock(&pTtlMgr->lock); + + return ret; +} diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index c3fb5e5ad4c3841ab48f8b86182882b19d543938..6c85488237ff6ae9b130e2d7d0070ea4698fd98e 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -37,7 +37,7 @@ static int32_t vnodeProcessCreateIndexReq(SVnode *pVnode, int64_t ver, void *pRe static int32_t vnodeProcessDropIndexReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); static int32_t vnodeProcessCompactVnodeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp); -static int32_t vnodePreprocessCreateTableReq(SVnode *pVnode, SDecoder *pCoder, int64_t ctime, int64_t *pUid) { +static int32_t vnodePreprocessCreateTableReq(SVnode *pVnode, SDecoder *pCoder, int64_t btime, int64_t *pUid) { int32_t code = 0; int32_t lino = 0; @@ -66,8 +66,8 @@ static int32_t vnodePreprocessCreateTableReq(SVnode *pVnode, SDecoder *pCoder, i } *(int64_t *)(pCoder->data + pCoder->pos) = uid; - // ctime - *(int64_t *)(pCoder->data + pCoder->pos + 8) = ctime; + // btime + *(int64_t *)(pCoder->data + pCoder->pos + 8) = btime; tEndDecode(pCoder); @@ -84,7 +84,7 @@ static int32_t vnodePreProcessCreateTableMsg(SVnode *pVnode, SRpcMsg *pMsg) { int32_t code = 0; int32_t lino = 0; - int64_t ctime = taosGetTimestampMs(); + int64_t btime = taosGetTimestampMs(); SDecoder dc = {0}; int32_t nReqs; @@ -99,7 +99,7 @@ static int32_t vnodePreProcessCreateTableMsg(SVnode *pVnode, SRpcMsg *pMsg) { TSDB_CHECK_CODE(code, lino, _exit); } for (int32_t iReq = 0; iReq < nReqs; iReq++) { - code = vnodePreprocessCreateTableReq(pVnode, &dc, ctime, NULL); + code = vnodePreprocessCreateTableReq(pVnode, &dc, btime, NULL); TSDB_CHECK_CODE(code, lino, _exit); } @@ -109,8 +109,35 @@ _exit: tDecoderClear(&dc); return code; } + +static int32_t vnodePreProcessAlterTableMsg(SVnode *pVnode, SRpcMsg *pMsg) { + int32_t code = TSDB_CODE_INVALID_MSG; + int32_t lino = 0; + + SDecoder dc = {0}; + tDecoderInit(&dc, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead)); + + SVAlterTbReq vAlterTbReq = {0}; + int64_t ctimeMs = taosGetTimestampMs(); + if (tDecodeSVAlterTbReqSetCtime(&dc, &vAlterTbReq, ctimeMs) < 0) { + goto _exit; + } + + code = 0; + +_exit: + tDecoderClear(&dc); + if (code) { + vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code)); + } else { + vTrace("vgId:%d %s done, table:%s ctimeMs generated:%" PRId64, TD_VID(pVnode), __func__, vAlterTbReq.tbName, + ctimeMs); + } + return code; +} + extern int64_t tsMaxKeyByPrecision[]; -static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int64_t ctime) { +static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int64_t btimeMs, int64_t ctimeMs) { int32_t code = 0; int32_t lino = 0; @@ -127,7 +154,7 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int int64_t uid; if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) { - code = vnodePreprocessCreateTableReq(pVnode, pCoder, ctime, &uid); + code = vnodePreprocessCreateTableReq(pVnode, pCoder, btimeMs, &uid); TSDB_CHECK_CODE(code, lino, _exit); } @@ -153,7 +180,7 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int } // scan and check - TSKEY now = ctime; + TSKEY now = btimeMs; if (pVnode->config.tsdbCfg.precision == TSDB_TIME_PRECISION_MICRO) { now *= 1000; } else if (pVnode->config.tsdbCfg.precision == TSDB_TIME_PRECISION_NANO) { @@ -170,7 +197,6 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int SColData colData = {0}; pCoder->pos += tGetColData(pCoder->data + pCoder->pos, &colData); - if (colData.flag != HAS_VALUE) { code = TSDB_CODE_INVALID_MSG; goto _exit; @@ -182,6 +208,10 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int goto _exit; } } + + for (uint64_t i = 1; i < nColData; i++) { + pCoder->pos += tGetColData(pCoder->data + pCoder->pos, &colData); + } } else { uint64_t nRow; if (tDecodeU64v(pCoder, &nRow) < 0) { @@ -200,6 +230,9 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int } } + *(int64_t *)(pCoder->data + pCoder->pos) = ctimeMs; + pCoder->pos += sizeof(int64_t); + tEndDecode(pCoder); _exit: @@ -229,9 +262,10 @@ static int32_t vnodePreProcessSubmitMsg(SVnode *pVnode, SRpcMsg *pMsg) { TSDB_CHECK_CODE(code, lino, _exit); } - int64_t ctime = taosGetTimestampMs(); + int64_t btimeMs = taosGetTimestampMs(); + int64_t ctimeMs = btimeMs; for (int32_t i = 0; i < nSubmitTbData; i++) { - code = vnodePreProcessSubmitTbData(pVnode, pCoder, ctime); + code = vnodePreProcessSubmitTbData(pVnode, pCoder, btimeMs, ctimeMs); TSDB_CHECK_CODE(code, lino, _exit); } @@ -257,6 +291,7 @@ static int32_t vnodePreProcessDeleteMsg(SVnode *pVnode, SRpcMsg *pMsg) { code = qWorkerProcessDeleteMsg(&handle, pVnode->pQuery, pMsg, &res); if (code) goto _exit; + res.ctimeMs = taosGetTimestampMs(); // malloc and encode tEncodeSize(tEncodeDeleteRes, &res, size, ret); pCont = rpcMallocCont(size + sizeof(SMsgHead)); @@ -278,6 +313,31 @@ _exit: return code; } +static int32_t vnodePreProcessBatchDeleteMsg(SVnode *pVnode, SRpcMsg *pMsg) { + int32_t code = 0; + int32_t lino = 0; + + int64_t ctimeMs = taosGetTimestampMs(); + SBatchDeleteReq pReq = {0}; + SDecoder *pCoder = &(SDecoder){0}; + + tDecoderInit(pCoder, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead)); + + if (tDecodeSBatchDeleteReqSetCtime(pCoder, &pReq, ctimeMs) < 0) { + code = TSDB_CODE_INVALID_MSG; + } + + tDecoderClear(pCoder); + taosArrayDestroy(pReq.deleteReqs); + + if (code) { + vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code)); + } else { + vTrace("vgId:%d %s done, ctimeMs generated:%" PRId64, TD_VID(pVnode), __func__, ctimeMs); + } + return code; +} + int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) { int32_t code = 0; @@ -285,12 +345,18 @@ int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) { case TDMT_VND_CREATE_TABLE: { code = vnodePreProcessCreateTableMsg(pVnode, pMsg); } break; + case TDMT_VND_ALTER_TABLE: { + code = vnodePreProcessAlterTableMsg(pVnode, pMsg); + } break; case TDMT_VND_SUBMIT: { code = vnodePreProcessSubmitMsg(pVnode, pMsg); } break; case TDMT_VND_DELETE: { code = vnodePreProcessDeleteMsg(pVnode, pMsg); } break; + case TDMT_VND_BATCH_DEL: { + code = vnodePreProcessBatchDeleteMsg(pVnode, pMsg); + } break; default: break; } @@ -637,8 +703,8 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t ver, void *pReq, goto end; } - vDebug("vgId:%d, drop ttl table req will be processed, time:%d", pVnode->config.vgId, ttlReq.timestamp); - int32_t ret = metaTtlDropTable(pVnode->pMeta, ttlReq.timestamp, tbUids); + vDebug("vgId:%d, drop ttl table req will be processed, time:%" PRId32, pVnode->config.vgId, ttlReq.timestampSec); + int32_t ret = metaTtlDropTable(pVnode->pMeta, (int64_t)ttlReq.timestampSec * 1000, tbUids); if (ret != 0) { goto end; } @@ -646,7 +712,7 @@ static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t ver, void *pReq, tqUpdateTbUidList(pVnode->pTq, tbUids, false); } - vnodeAsyncRentention(pVnode, ttlReq.timestamp); + vnodeAsyncRentention(pVnode, ttlReq.timestampSec); end: taosArrayDestroy(tbUids); @@ -1382,6 +1448,9 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, in code = tsdbInsertTableData(pVnode->pTsdb, ver, pSubmitTbData, &affectedRows); if (code) goto _exit; + code = metaUpdateChangeTime(pVnode->pMeta, pSubmitTbData->uid, pSubmitTbData->ctimeMs); + if (code) goto _exit; + pSubmitRsp->affectedRows += affectedRows; } @@ -1636,6 +1705,14 @@ static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pRe TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs); } + code = metaUpdateChangeTime(pVnode->pMeta, uid, deleteReq.ctimeMs); + if (code < 0) { + terrno = code; + vError("vgId:%d, update change time error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64 + ", end ts:%" PRId64, + TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs); + } + tDecoderClear(&mr.coder); } metaReaderClear(&mr); @@ -1664,8 +1741,10 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, in ASSERT(taosArrayGetSize(pRes->uidList) == 0 || (pRes->skey != 0 && pRes->ekey != 0)); for (int32_t iUid = 0; iUid < taosArrayGetSize(pRes->uidList); iUid++) { - code = tsdbDeleteTableData(pVnode->pTsdb, ver, pRes->suid, *(uint64_t *)taosArrayGet(pRes->uidList, iUid), - pRes->skey, pRes->ekey); + uint64_t uid = *(uint64_t *)taosArrayGet(pRes->uidList, iUid); + code = tsdbDeleteTableData(pVnode->pTsdb, ver, pRes->suid, uid, pRes->skey, pRes->ekey); + if (code) goto _err; + code = metaUpdateChangeTime(pVnode->pMeta, uid, pRes->ctimeMs); if (code) goto _err; } diff --git a/source/libs/executor/src/sysscanoperator.c b/source/libs/executor/src/sysscanoperator.c index 23a7d2c9e9e0f8c6415d2a8cccb53e5b7ee0cfd4..06138d7d5c5bd8d438451b07198f85c793e02f8b 100644 --- a/source/libs/executor/src/sysscanoperator.c +++ b/source/libs/executor/src/sysscanoperator.c @@ -1154,7 +1154,7 @@ static SSDataBlock* sysTableBuildUserTablesByUids(SOperatorInfo* pOperator) { int32_t tableType = mr.me.type; if (tableType == TSDB_CHILD_TABLE) { // create time - int64_t ts = mr.me.ctbEntry.ctime; + int64_t ts = mr.me.ctbEntry.btime; pColInfoData = taosArrayGet(p->pDataBlock, 2); colDataSetVal(pColInfoData, numOfRows, (char*)&ts, false); @@ -1206,7 +1206,7 @@ static SSDataBlock* sysTableBuildUserTablesByUids(SOperatorInfo* pOperator) { } else if (tableType == TSDB_NORMAL_TABLE) { // create time pColInfoData = taosArrayGet(p->pDataBlock, 2); - colDataSetVal(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.ctime, false); + colDataSetVal(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.btime, false); // number of columns pColInfoData = taosArrayGet(p->pDataBlock, 3); @@ -1338,7 +1338,7 @@ static SSDataBlock* sysTableBuildUserTables(SOperatorInfo* pOperator) { int32_t tableType = pInfo->pCur->mr.me.type; if (tableType == TSDB_CHILD_TABLE) { // create time - int64_t ts = pInfo->pCur->mr.me.ctbEntry.ctime; + int64_t ts = pInfo->pCur->mr.me.ctbEntry.btime; pColInfoData = taosArrayGet(p->pDataBlock, 2); colDataSetVal(pColInfoData, numOfRows, (char*)&ts, false); @@ -1392,7 +1392,7 @@ static SSDataBlock* sysTableBuildUserTables(SOperatorInfo* pOperator) { } else if (tableType == TSDB_NORMAL_TABLE) { // create time pColInfoData = taosArrayGet(p->pDataBlock, 2); - colDataSetVal(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.ctime, false); + colDataSetVal(pColInfoData, numOfRows, (char*)&pInfo->pCur->mr.me.ntbEntry.btime, false); // number of columns pColInfoData = taosArrayGet(p->pDataBlock, 3); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 5890ba15f4131273cfa92dfe958e87aeb3f6a04f..9bcdd102a6f2b7fdddec53077938d86303f91f1c 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2335,7 +2335,7 @@ static int32_t checkHavingGroupBy(STranslateContext* pCxt, SSelectStmt* pSelect) int32_t code = TSDB_CODE_SUCCESS; if (NULL == getGroupByList(pCxt) && NULL == pSelect->pPartitionByList && NULL == pSelect->pWindow) { return code; - } + } if (NULL != pSelect->pHaving) { code = checkExprForGroupBy(pCxt, &pSelect->pHaving); } @@ -2346,7 +2346,7 @@ static int32_t checkHavingGroupBy(STranslateContext* pCxt, SSelectStmt* pSelect) if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pOrderByList) { code = checkExprListForGroupBy(pCxt, pSelect, pSelect->pOrderByList); } -*/ +*/ return code; } @@ -2654,7 +2654,7 @@ static int32_t replaceTbName(STranslateContext* pCxt, SSelectStmt* pSelect) { SNode** pNode = NULL; SRewriteTbNameContext pRewriteCxt = {0}; pRewriteCxt.pTbName = pTable->table.tableName; - + nodesRewriteExprPostOrder(&pSelect->pWhere, doTranslateTbName, &pRewriteCxt); return pRewriteCxt.errCode; @@ -3151,7 +3151,7 @@ static int32_t translateSelectList(STranslateContext* pCxt, SSelectStmt* pSelect } if (TSDB_CODE_SUCCESS == code) { code = checkExprListForGroupBy(pCxt, pSelect, pSelect->pProjectionList); - } + } if (TSDB_CODE_SUCCESS == code) { code = translateFillValues(pCxt, pSelect); } @@ -4939,6 +4939,14 @@ static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt if (TSDB_CODE_SUCCESS == code) { code = checkTableSchema(pCxt, pStmt); } + if (TSDB_CODE_SUCCESS == code) { + if (createStable && pStmt->pOptions->ttl != 0) { + code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_TABLE_OPTION, + "Only supported for create non-super table in databases " + "configured with the 'TTL' option"); + } + } + return code; } diff --git a/source/libs/parser/test/parInitialCTest.cpp b/source/libs/parser/test/parInitialCTest.cpp index a4e8bdd87a6072a20722e0c858775261f8c593b0..67767f71f1181ac6e20c277e4d41b587c9f9d4a3 100644 --- a/source/libs/parser/test/parInitialCTest.cpp +++ b/source/libs/parser/test/parInitialCTest.cpp @@ -765,7 +765,7 @@ TEST_F(ParserInitialCTest, createStable) { "TAGS (a1 TIMESTAMP, a2 INT, a3 INT UNSIGNED, a4 BIGINT, a5 BIGINT UNSIGNED, a6 FLOAT, a7 DOUBLE, " "a8 BINARY(20), a9 SMALLINT, a10 SMALLINT UNSIGNED, a11 TINYINT, " "a12 TINYINT UNSIGNED, a13 BOOL, a14 NCHAR(30), a15 VARCHAR(50)) " - "TTL 100 COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN) MAX_DELAY 100s,10m WATERMARK 10a,1m " + "COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN) MAX_DELAY 100s,10m WATERMARK 10a,1m " "DELETE_MARK 1000s,200m"); clearCreateStbReq(); } @@ -1005,7 +1005,7 @@ TEST_F(ParserInitialCTest, createTable) { ASSERT_EQ(pReq->flags, pExpect->flags); ASSERT_EQ(std::string(pReq->name), std::string(pExpect->name)); ASSERT_EQ(pReq->uid, pExpect->uid); - ASSERT_EQ(pReq->ctime, pExpect->ctime); + ASSERT_EQ(pReq->btime, pExpect->btime); ASSERT_EQ(pReq->ttl, pExpect->ttl); ASSERT_EQ(pReq->commentLen, pExpect->commentLen); ASSERT_EQ(std::string(pReq->comment), std::string(pExpect->comment)); @@ -1038,7 +1038,7 @@ TEST_F(ParserInitialCTest, createTable) { "TAGS (a1 TIMESTAMP, a2 INT, a3 INT UNSIGNED, a4 BIGINT, a5 BIGINT UNSIGNED, a6 FLOAT, a7 DOUBLE, a8 BINARY(20), " "a9 SMALLINT, a10 SMALLINT UNSIGNED, a11 TINYINT, a12 TINYINT UNSIGNED, a13 BOOL, " "a14 NCHAR(30), a15 VARCHAR(50)) " - "TTL 100 COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN)"); + "COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN)"); run("CREATE TABLE IF NOT EXISTS t1 USING st1 TAGS(1, 'wxy', NOW)"); diff --git a/source/libs/qcom/src/queryUtil.c b/source/libs/qcom/src/queryUtil.c index fa2ddd9163c8309c3853e5c189dd1f9124c64307..38d7c9da3bb5a01848c2055d28a982272173a289 100644 --- a/source/libs/qcom/src/queryUtil.c +++ b/source/libs/qcom/src/queryUtil.c @@ -141,7 +141,7 @@ void destroySendMsgInfo(SMsgSendInfo* pMsgBody) { if (NULL == pMsgBody) { return; } - + taosMemoryFreeClear(pMsgBody->target.dbFName); taosMemoryFreeClear(pMsgBody->msgInfo.pData); if (pMsgBody->paramFreeFp) { @@ -173,7 +173,7 @@ int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTra .contLen = pInfo->msgInfo.len, .info.ahandle = (void*)pInfo, .info.handle = pInfo->msgInfo.handle, - .info.persistHandle = persistHandle, + .info.persistHandle = persistHandle, .code = 0 }; TRACE_SET_ROOTID(&rpcMsg.info.traceId, pInfo->requestId); @@ -252,7 +252,7 @@ void destroyQueryExecRes(SExecResult* pRes) { taosMemoryFreeClear(pRes->res); break; } - case TDMT_SCH_QUERY: + case TDMT_SCH_QUERY: case TDMT_SCH_MERGE_QUERY: { taosArrayDestroy((SArray*)pRes->res); break; @@ -532,7 +532,7 @@ int32_t cloneSVreateTbReq(SVCreateTbReq* pSrc, SVCreateTbReq** pDst) { (*pDst)->name = taosStrdup(pSrc->name); } (*pDst)->uid = pSrc->uid; - (*pDst)->ctime = pSrc->ctime; + (*pDst)->btime = pSrc->btime; (*pDst)->ttl = pSrc->ttl; (*pDst)->commentLen = pSrc->commentLen; if (pSrc->comment) { diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index 9db049508126d2665726f79a56456b2c09e45656..cd6794a7ee5824082bff600b63b06df5dab41f62 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -427,6 +427,7 @@ int32_t qwGetDeleteResFromSink(QW_FPARAMS_DEF, SQWTaskCtx *ctx, SDeleteRes *pRes pRes->suid = pDelRes->suid; pRes->uidList = pDelRes->uidList; + pRes->ctimeMs = taosGetTimestampMs(); pRes->skey = pDelRes->skey; pRes->ekey = pDelRes->ekey; pRes->affectedRows = pDelRes->affectedRows; diff --git a/source/libs/tdb/inc/tdb.h b/source/libs/tdb/inc/tdb.h index 0e20941b3ae472986d5116616eda05af93a83dc1..4dd47e97ac94bdba2b80bd6f5815a38e987bc9d8 100644 --- a/source/libs/tdb/inc/tdb.h +++ b/source/libs/tdb/inc/tdb.h @@ -46,12 +46,16 @@ int32_t tdbAlter(TDB *pDb, int pages); int32_t tdbTbOpen(const char *tbname, int keyLen, int valLen, tdb_cmpr_fn_t keyCmprFn, TDB *pEnv, TTB **ppTb, int8_t rollback); int32_t tdbTbClose(TTB *pTb); +bool tdbTbExist(const char *tbname, TDB *pEnv); +int tdbTbDropByName(const char *tbname, TDB *pEnv, TXN* pTxn); int32_t tdbTbDrop(TTB *pTb); int32_t tdbTbInsert(TTB *pTb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn); int32_t tdbTbDelete(TTB *pTb, const void *pKey, int kLen, TXN *pTxn); int32_t tdbTbUpsert(TTB *pTb, const void *pKey, int kLen, const void *pVal, int vLen, TXN *pTxn); int32_t tdbTbGet(TTB *pTb, const void *pKey, int kLen, void **ppVal, int *vLen); int32_t tdbTbPGet(TTB *pTb, const void *pKey, int kLen, void **ppKey, int *pkLen, void **ppVal, int *vLen); +int32_t tdbTbTraversal(TTB *pTb, void *data, + int32_t (*func)(const void *pKey, int keyLen, const void *pVal, int valLen, void *data)); // TBC int32_t tdbTbcOpen(TTB *pTb, TBC **ppTbc, TXN *pTxn); diff --git a/source/libs/tdb/src/db/tdbTable.c b/source/libs/tdb/src/db/tdbTable.c index 18d14fa474f51104442cf689bd52649598b9c396..446a21f312ba4e3118744ae2f8837126c9ff3fcf 100644 --- a/source/libs/tdb/src/db/tdbTable.c +++ b/source/libs/tdb/src/db/tdbTable.c @@ -134,11 +134,67 @@ int tdbTbClose(TTB *pTb) { return 0; } +bool tdbTbExist(const char *tbname, TDB *pEnv) { + bool exist = false; + SPager *pPager; + char fFullName[TDB_FILENAME_LEN]; + +#ifdef USE_MAINDB + + snprintf(fFullName, TDB_FILENAME_LEN, "%s/%s", pEnv->dbName, TDB_MAINDB_NAME); + + if (strcmp(TDB_MAINDB_NAME, tbname)) { + pPager = tdbEnvGetPager(pEnv, fFullName); + + exist = tdbTbGet(pPager->pEnv->pMainDb, tbname, strlen(tbname) + 1, NULL, NULL) == 0; + } else { + exist = taosCheckExistFile(fFullName); + } + +#else + + snprintf(fFullName, TDB_FILENAME_LEN, "%s/%s", pEnv->dbName, tbname); + + exist = taosCheckExistFile(fFullName); + +#endif + + return exist; +} + int tdbTbDrop(TTB *pTb) { // TODO return 0; } +int tdbTbDropByName(const char *tbname, TDB *pEnv, TXN *pTxn) { + int ret; + SPager *pPager; + char fFullName[TDB_FILENAME_LEN]; + +#ifdef USE_MAINDB + + snprintf(fFullName, TDB_FILENAME_LEN, "%s/%s", pEnv->dbName, TDB_MAINDB_NAME); + + if (strcmp(TDB_MAINDB_NAME, tbname)) { + pPager = tdbEnvGetPager(pEnv, fFullName); + + ret = tdbTbDelete(pPager->pEnv->pMainDb, tbname, strlen(tbname) + 1, pTxn); + } else { + ret = taosRemoveFile(fFullName); + } + +#else + + snprintf(fFullName, TDB_FILENAME_LEN, "%s/%s", pEnv->dbName, tbname); + + ret = taosRemoveFile(fFullName); + +#endif + + return ret; +} + int tdbTbInsert(TTB *pTb, const void *pKey, int keyLen, const void *pVal, int valLen, TXN *pTxn) { return tdbBtreeInsert(pTb->pBt, pKey, keyLen, pVal, valLen, pTxn); } @@ -173,6 +229,38 @@ int tdbTbcOpen(TTB *pTb, TBC **ppTbc, TXN *pTxn) { return 0; } +int32_t tdbTbTraversal(TTB *pTb, void *data, + int32_t (*func)(const void *pKey, int keyLen, const void *pVal, int valLen, void *data)) { + TBC *pCur; + int ret = tdbTbcOpen(pTb, &pCur, NULL); + if (ret < 0) { + return ret; + } + + tdbTbcMoveToFirst(pCur); + + void *pKey = NULL; + int kLen = 0; + void *pValue = NULL; + int vLen = 0; + + while (1) { + ret = tdbTbcNext(pCur, &pKey, &kLen, &pValue, &vLen); + if (ret < 0) { + ret = 0; + break; + } + + ret = func(pKey, kLen, pValue, vLen, data); + if (ret < 0) break; + } + tdbFree(pKey); + tdbFree(pValue); + tdbTbcClose(pCur); + + return 0; +} + int tdbTbcMoveTo(TBC *pTbc, const void *pKey, int kLen, int *c) { return tdbBtcMoveTo(&pTbc->btc, pKey, kLen, c); } int tdbTbcMoveToFirst(TBC *pTbc) { return tdbBtcMoveToFirst(&pTbc->btc); } diff --git a/tests/system-test/1-insert/table_param_ttl.py b/tests/system-test/1-insert/table_param_ttl.py index 51d7d596cd9d94e6078fc7d448583c91a27e1780..6cc978a76ce6f71e2dc0b68fc5561bde7e9a6656 100644 --- a/tests/system-test/1-insert/table_param_ttl.py +++ b/tests/system-test/1-insert/table_param_ttl.py @@ -35,14 +35,14 @@ class TDTestCase: tdSql.execute(f'create table db.{self.ntbname}_{i} (ts timestamp,c0 int) ttl {self.ttl_param}') tdSql.query(f'show db.tables') tdSql.checkRows(self.tbnum) - sleep(self.updatecfgDict['ttlUnit']*self.ttl_param+self.updatecfgDict['ttlPushInterval']) + sleep(self.updatecfgDict['ttlUnit']*self.ttl_param+self.updatecfgDict['ttlPushInterval'] + 1) tdSql.query(f'show db.tables') tdSql.checkRows(0) for i in range(self.tbnum): tdSql.execute(f'create table db.{self.ntbname}_{i} (ts timestamp,c0 int) ttl {self.default_ttl}') for i in range(int(self.tbnum/2)): tdSql.execute(f'alter table db.{self.ntbname}_{i} ttl {self.modify_ttl}') - sleep(self.updatecfgDict['ttlUnit']*self.modify_ttl+self.updatecfgDict['ttlPushInterval']) + sleep(self.updatecfgDict['ttlUnit']*self.modify_ttl+self.updatecfgDict['ttlPushInterval'] + 1) tdSql.query(f'show db.tables') tdSql.checkRows(self.tbnum - int(self.tbnum/2)) tdSql.execute('drop database db') @@ -54,7 +54,7 @@ class TDTestCase: tdSql.execute(f'create table db.{self.stbname}_{i} using db.{self.stbname} tags({i}) ttl {self.ttl_param}') tdSql.query(f'show db.tables') tdSql.checkRows(self.tbnum) - sleep(self.updatecfgDict['ttlUnit']*self.ttl_param+self.updatecfgDict['ttlPushInterval']) + sleep(self.updatecfgDict['ttlUnit']*self.ttl_param+self.updatecfgDict['ttlPushInterval'] + 1) tdSql.query(f'show db.tables') tdSql.checkRows(0) for i in range(self.tbnum): @@ -63,7 +63,7 @@ class TDTestCase: tdSql.checkRows(self.tbnum) for i in range(int(self.tbnum/2)): tdSql.execute(f'alter table db.{self.stbname}_{i} ttl {self.modify_ttl}') - sleep(self.updatecfgDict['ttlUnit']*self.modify_ttl+self.updatecfgDict['ttlPushInterval']) + sleep(self.updatecfgDict['ttlUnit']*self.modify_ttl+self.updatecfgDict['ttlPushInterval'] + 1) tdSql.query(f'show db.tables') tdSql.checkRows(self.tbnum - int(self.tbnum/2)) tdSql.execute('drop database db') @@ -75,7 +75,7 @@ class TDTestCase: tdSql.execute(f'insert into db.{self.stbname}_{i} using db.{self.stbname} tags({i}) ttl {self.ttl_param} values(now,1)') tdSql.query(f'show db.tables') tdSql.checkRows(self.tbnum) - sleep(self.updatecfgDict['ttlUnit']*self.ttl_param+self.updatecfgDict['ttlPushInterval']) + sleep(self.updatecfgDict['ttlUnit']*self.ttl_param+self.updatecfgDict['ttlPushInterval'] + 1) tdSql.query(f'show db.tables') tdSql.checkRows(0) tdSql.execute('drop database db') diff --git a/utils/test/c/write_raw_block_test.c b/utils/test/c/write_raw_block_test.c index 8e5dd6275217b194fa4154fdd1235fdb27e397f1..ee2594af7a89a5710336177debcaff62d6936a59 100644 --- a/utils/test/c/write_raw_block_test.c +++ b/utils/test/c/write_raw_block_test.c @@ -52,7 +52,7 @@ int buildStable(TAOS* pConn, TAOS_RES* pRes) { pRes = taos_query(pConn, "create table d2 using meters tags(3, 'San Francisco')"); if (taos_errno(pRes) != 0) { - printf("failed to create child table d1, reason:%s\n", taos_errstr(pRes)); + printf("failed to create child table d2, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes);