diff --git a/docs-cn/20-third-party/01-grafana.mdx b/docs-cn/20-third-party/01-grafana.mdx index 31cd434080e4fea56d1817a25fba5078b223e164..09c0d786cf5ac3fc9a6c5e6a72087be0b658454d 100644 --- a/docs-cn/20-third-party/01-grafana.mdx +++ b/docs-cn/20-third-party/01-grafana.mdx @@ -38,6 +38,16 @@ export TDENGINE_API=http://tdengine.local:6041 # user + password export TDENGINE_USER=user export TDENGINE_PASSWORD=password + +# 其他环境变量： +# - 是否安装数据源，默认为 true，表示安装 +export TDENGINE_DS_ENABLED=false +# - 数据源名称，默认为 TDengine +export TDENGINE_DS_NAME=TDengine +# - 数据源所属组织 ID，默认为 1 +export GF_ORG_ID=1 +# - 数据源是否可通过管理面板编辑，默认为 0，表示不可编辑 +export TDENGINE_EDITABLE=1 ``` 运行安装脚本： @@ -48,6 +58,8 @@ bash -c "$(curl -fsSL https://raw.githubusercontent.com/taosdata/grafanaplugin/m 该脚本将自动安装 Grafana 插件并配置数据源。安装完毕后，需要重启 Grafana 服务后生效。 +保存该脚本并执行 `./install.sh --help` 可查看详细帮助文档。 + diff --git a/docs-en/20-third-party/01-grafana.mdx b/docs-en/20-third-party/01-grafana.mdx index 033ef3b20692966863b2d8b5e61f57e7da38732e..9076f163bea60e87b5d1aaed79d966572463d56d 100644 --- a/docs-en/20-third-party/01-grafana.mdx +++ b/docs-en/20-third-party/01-grafana.mdx @@ -40,6 +40,16 @@ export TDENGINE_API=http://tdengine.local:6041 # user + password export TDENGINE_USER=user export TDENGINE_PASSWORD=password + +# Other useful variables +# - If to install TDengine data source, default is true +export TDENGINE_DS_ENABLED=false +# - Data source name to be created, default is TDengine +export TDENGINE_DS_NAME=TDengine +# - Data source organization id, default is 1 +export GF_ORG_ID=1 +# - Data source is editable in admin ui or not, default is 0 (false) +export TDENGINE_EDITABLE=1 ``` Run `install.sh`: @@ -48,7 +58,7 @@ Run `install.sh`: bash -c "$(curl -fsSL https://raw.githubusercontent.com/taosdata/grafanaplugin/master/install.sh)" ``` -With this script, TDengine data source plugin and the Grafana data source will be installed and created automatically with Grafana provisioning configurations. +With this script, TDengine data source plugin and the Grafana data source will be installed and created automatically with Grafana provisioning configurations. Save the script and type `./install.sh --help` for the full usage of the script. And then, restart Grafana service and open Grafana in web-browser, usually . diff --git a/include/common/tcommon.h b/include/common/tcommon.h index cc51f96f6c7c0a5fed3b231a360335ad92429cf8..2e646f4769607947102be1dbae767410a167a6e8 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -37,14 +37,6 @@ enum { TMQ_MSG_TYPE__EP_RSP, }; -enum { - STREAM_TRIGGER__AT_ONCE = 1, - STREAM_TRIGGER__WINDOW_CLOSE, - STREAM_TRIGGER__BY_COUNT, - STREAM_TRIGGER__BY_BATCH_COUNT, - STREAM_TRIGGER__BY_EVENT_TIME, -}; - typedef enum EStreamType { STREAM_NORMAL = 1, STREAM_INVERT, diff --git a/include/util/tlrucache.h b/include/util/tlrucache.h new file mode 100644 index 0000000000000000000000000000000000000000..5aee50c42abe3bf41972231df07e8cabe67a178e --- /dev/null +++ b/include/util/tlrucache.h @@ -0,0 +1,71 @@ +/* + * 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_UTIL_LRUCACHE_H_ +#define _TD_UTIL_LRUCACHE_H_ + +#include "thash.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct SLRUCache SLRUCache; + +typedef void (*_taos_lru_deleter_t)(const void *key, size_t keyLen, void *value); + +typedef struct LRUHandle LRUHandle; + +typedef enum { + TAOS_LRU_PRIORITY_HIGH, + TAOS_LRU_PRIORITY_LOW +} LRUPriority; + +typedef enum { + TAOS_LRU_STATUS_OK, + TAOS_LRU_STATUS_FAIL, + TAOS_LRU_STATUS_INCOMPLETE, + TAOS_LRU_STATUS_OK_OVERWRITTEN +} LRUStatus; + +SLRUCache *taosLRUCacheInit(size_t capacity, int numShardBits, double highPriPoolRatio); +void taosLRUCacheCleanup(SLRUCache *cache); + +LRUStatus taosLRUCacheInsert(SLRUCache *cache, const void *key, size_t keyLen, void *value, size_t charge, + _taos_lru_deleter_t deleter, LRUHandle **handle, LRUPriority priority); +LRUHandle *taosLRUCacheLookup(SLRUCache * cache, const void *key, size_t keyLen); +void taosLRUCacheErase(SLRUCache * cache, const void *key, size_t keyLen); + +void taosLRUCacheEraseUnrefEntries(SLRUCache *cache); + +bool taosLRUCacheRef(SLRUCache *cache, LRUHandle *handle); +bool taosLRUCacheRelease(SLRUCache *cache, LRUHandle *handle, bool eraseIfLastRef); + +void* taosLRUCacheValue(SLRUCache *cache, LRUHandle *handle); + +size_t taosLRUCacheGetUsage(SLRUCache *cache); +size_t taosLRUCacheGetPinnedUsage(SLRUCache *cache); + +void taosLRUCacheSetCapacity(SLRUCache *cache, size_t capacity); +size_t taosLRUCacheGetCapacity(SLRUCache *cache); + +void taosLRUCacheSetStrictCapacity(SLRUCache *cache, bool strict); +bool taosLRUCacheIsStrictCapacity(SLRUCache *cache); + +#ifdef __cplusplus +} +#endif + +#endif // _TD_UTIL_LRUCACHE_H_ diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index 171f06c2575af50ca7b514c70b9b30ddb601b46b..bb60624145f47545769b3a606a25db93268ac1ce 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -167,7 +167,7 @@ void *createTscObj(const char *user, const char *auth, const char *db, int32_t c taosThreadMutexInit(&pObj->mutex, NULL); pObj->id = taosAddRef(clientConnRefPool, pObj); - pObj->schemalessType = 0; + pObj->schemalessType = 1; tscDebug("connObj created, 0x%" PRIx64, pObj->id); return pObj; diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index 68bfa99eadfde94f1b86e90e21d41ec419a2f457..bbf9eba363a398c243439f811e9046c796fa027c 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -306,19 +306,10 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) { const char* errStr = taos_errstr(res); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%"PRIx64" apply schema action. error: %s", info->id, errStr); + taosMsleep(100); } taos_free_result(res); -// if (code == TSDB_CODE_MND_FIELD_ALREADY_EXIST || code == TSDB_CODE_MND_TAG_ALREADY_EXIST || tscDupColNames) { - if (code == TSDB_CODE_MND_TAG_ALREADY_EXIST) { - TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE"); - code = taos_errno(res2); - if (code != TSDB_CODE_SUCCESS) { - uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2)); - } - taos_free_result(res2); - taosMsleep(500); - } break; } case SCHEMA_ACTION_ADD_TAG: { @@ -330,19 +321,10 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) { const char* errStr = taos_errstr(res); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res)); + taosMsleep(100); } taos_free_result(res); -// if (code ==TSDB_CODE_MND_TAG_ALREADY_EXIST || code == TSDB_CODE_MND_FIELD_ALREAY_EXIST || tscDupColNames) { - if (code ==TSDB_CODE_MND_TAG_ALREADY_EXIST) { - TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE"); - code = taos_errno(res2); - if (code != TSDB_CODE_SUCCESS) { - uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2)); - } - taos_free_result(res2); - taosMsleep(500); - } break; } case SCHEMA_ACTION_CHANGE_COLUMN_SIZE: { @@ -353,19 +335,10 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) { code = taos_errno(res); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res)); + taosMsleep(100); } taos_free_result(res); -// if (code == TSDB_CODE_MND_INVALID_COLUMN_LENGTH || code == TSDB_CODE_TSC_INVALID_COLUMN_LENGTH) { - if (code == TSDB_CODE_TSC_INVALID_COLUMN_LENGTH) { - TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE"); - code = taos_errno(res2); - if (code != TSDB_CODE_SUCCESS) { - uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2)); - } - taos_free_result(res2); - taosMsleep(500); - } break; } case SCHEMA_ACTION_CHANGE_TAG_SIZE: { @@ -376,19 +349,10 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) { code = taos_errno(res); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res)); + taosMsleep(100); } taos_free_result(res); -// if (code == TSDB_CODE_MND_INVALID_TAG_LENGTH || code == TSDB_CODE_TSC_INVALID_TAG_LENGTH) { - if (code == TSDB_CODE_TSC_INVALID_TAG_LENGTH) { - TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE"); - code = taos_errno(res2); - if (code != TSDB_CODE_SUCCESS) { - uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2)); - } - taos_free_result(res2); - taosMsleep(500); - } break; } case SCHEMA_ACTION_CREATE_STABLE: { @@ -428,18 +392,10 @@ static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) { code = taos_errno(res); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res)); + taosMsleep(100); } taos_free_result(res); - if (code == TSDB_CODE_MND_STB_ALREADY_EXIST) { - TAOS_RES* res2 = taos_query(info->taos, "RESET QUERY CACHE"); - code = taos_errno(res2); - if (code != TSDB_CODE_SUCCESS) { - uError("SML:0x%" PRIx64 " apply schema action. reset query cache. error: %s", info->id, taos_errstr(res2)); - } - taos_free_result(res2); - taosMsleep(500); - } break; } @@ -473,6 +429,21 @@ static int32_t smlProcessSchemaAction(SSmlHandle* info, SSchema* schemaField, SH return TSDB_CODE_SUCCESS; } +static int32_t smlCheckMeta(SSchema* schema, int32_t length, SArray* cols){ + SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK); + for(uint16_t i = 0; i < length; i++){ + taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES); + } + + for(int32_t i = 0; i < taosArrayGetSize(cols); i++){ + SSmlKv* kv = (SSmlKv*)taosArrayGetP(cols, i); + if(taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL){ + return -1; + } + } + return 0; +} + static int32_t smlModifyDBSchemas(SSmlHandle* info) { int32_t code = 0; SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp); @@ -483,6 +454,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) { while (tableMetaSml) { SSmlSTableMeta* sTableData = *tableMetaSml; STableMeta *pTableMeta = NULL; + bool needCheckMeta = false; // for multi thread size_t superTableLen = 0; void *superTable = taosHashGetKey(tableMetaSml, &superTableLen); @@ -533,6 +505,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) { if (code != TSDB_CODE_SUCCESS) { goto end; } + needCheckMeta = true; } else { uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code)); goto end; @@ -544,6 +517,20 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) { uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, (char*)superTable); goto end; } + + if(needCheckMeta){ + code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags, sTableData->tags); + if (code != TSDB_CODE_SUCCESS) { + uError("SML:0x%"PRIx64" check tag failed. super table name %s", info->id, (char*)superTable); + goto end; + } + code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols); + if (code != TSDB_CODE_SUCCESS) { + uError("SML:0x%"PRIx64" check cols failed. super table name %s", info->id, (char*)superTable); + goto end; + } + } + sTableData->tableMeta = pTableMeta; tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, tableMetaSml); @@ -2368,6 +2355,7 @@ static void smlInsertCallback(void* param, void* res, int32_t code) { SRequestObj *pRequest = (SRequestObj *)res; SSmlHandle* info = (SSmlHandle *)param; + uDebug("SML:0x%"PRIx64" result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf); // lock if(code != TSDB_CODE_SUCCESS){ taosThreadSpinLock(&info->params->lock); @@ -2496,8 +2484,9 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr end: taosThreadSpinDestroy(¶ms.lock); tsem_destroy(¶ms.sem); - ((STscObj *)taos)->schemalessType = 0; - uDebug("result:%s", request->msgBuf); +// ((STscObj *)taos)->schemalessType = 0; + ((STscObj *)taos)->schemalessType = 1; + uDebug("resultend:%s", request->msgBuf); return (TAOS_RES*)request; } diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index 58905cac19e852a130cf7472f09b392a603a930f..7d49c4206f9c300f15dba0f09e24d37918fc8237 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -408,7 +408,7 @@ int32_t tmqCommitInner(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, int8_ pParam->userParam = userParam; if (!async) tsem_init(&pParam->rspSem, 0, 0); - sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo)); + sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) goto END; sendInfo->msgInfo = (SDataBuf){ .pData = buf, @@ -704,7 +704,7 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) { void* abuf = buf; tSerializeSCMSubscribeReq(&abuf, &req); - SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo)); + SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) goto FAIL; SMqSubscribeCbParam param = { @@ -1008,7 +1008,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) { pParam->async = async; tsem_init(&pParam->rspSem, 0, 0); - SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo)); + SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) { tsem_destroy(&pParam->rspSem); taosMemoryFree(pParam); @@ -1162,7 +1162,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) { pParam->vgId = pVg->vgId; pParam->epoch = tmq->epoch; - SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo)); + SMsgSendInfo* sendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); if (sendInfo == NULL) { taosMemoryFree(pReq); taosMemoryFree(pParam); diff --git a/source/client/test/CMakeLists.txt b/source/client/test/CMakeLists.txt index bb9f0ed23ee9ec5fb1b961f8d15370764814102b..03d2b4813446e97e7427f98356d7279bae355412 100644 --- a/source/client/test/CMakeLists.txt +++ b/source/client/test/CMakeLists.txt @@ -41,7 +41,7 @@ TARGET_INCLUDE_DIRECTORIES( PRIVATE "${TD_SOURCE_DIR}/source/client/inc" ) -#add_test( -# NAME smlTest -# COMMAND smlTest -#) +add_test( + NAME smlTest + COMMAND smlTest +) diff --git a/source/client/test/smlTest.cpp b/source/client/test/smlTest.cpp index 49e26a818f3d23bfe26076a9d71384cb83447c79..4ad73a6424b89aec1ab81eb1b29e7fb0e3dbe248 100644 --- a/source/client/test/smlTest.cpp +++ b/source/client/test/smlTest.cpp @@ -499,6 +499,7 @@ TEST(testCase, smlGetTimestampLen_Test) { ASSERT_EQ(len, 3); } +/* TEST(testCase, smlProcess_influx_Test) { TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0); ASSERT_NE(taos, nullptr); @@ -1259,4 +1260,4 @@ TEST(testCase, sml_16368_Test) { pRes = taos_schemaless_insert(taos, (char**)sql, sizeof(sql)/sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL, TSDB_SML_TIMESTAMP_MICRO_SECONDS); ASSERT_EQ(taos_errno(pRes), 0); taos_free_result(pRes); -} +}*/ diff --git a/source/common/src/systable.c b/source/common/src/systable.c index cb38a3cf70b163a2218640c46f7007e0000f03cb..08977abd61aeefc64ffca9702c6a04b7cd529858 100644 --- a/source/common/src/systable.c +++ b/source/common/src/systable.c @@ -91,7 +91,7 @@ static const SSysDbTableSchema userDBSchema[] = { {.name = "precision", .bytes = 2 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "single_stable_model", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL}, {.name = "status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, - {.name = "schemaless", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL}, +// {.name = "schemaless", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL}, {.name = "retension", .bytes = 60 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, // {.name = "update", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, // disable update @@ -175,7 +175,7 @@ static const SSysDbTableSchema userUsersSchema[] = { }; static const SSysDbTableSchema grantsSchema[] = { - {.name = "version", .bytes = 8 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + {.name = "version", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "expire time", .bytes = 19 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "expired", .bytes = 5 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "storage(GB)", .bytes = 21 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, diff --git a/source/dnode/mnode/impl/inc/mndGrant.h b/source/dnode/mnode/impl/inc/mndGrant.h index ad3dc7f79d2d9925e223880bde05560f1587fb11..525dd2c2e5184d03fb38d5bf9cdaaaa08f20709d 100644 --- a/source/dnode/mnode/impl/inc/mndGrant.h +++ b/source/dnode/mnode/impl/inc/mndGrant.h @@ -20,6 +20,8 @@ "C" { #endif +#include "mndInt.h" + typedef enum { TSDB_GRANT_ALL, TSDB_GRANT_TIME, diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index c20459829ece38be56beb6d6fcf9b27f1684577a..b4237466d4ca84442e0a0e4ef0c5b1dbcbf0da69 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -1503,8 +1503,8 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, rows, (const char *)statusB, false); - pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); - colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.schemaless, false); +// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); +// colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.schemaless, false); char *p = buildRetension(pDb->cfg.pRetensions); diff --git a/source/dnode/mnode/impl/src/mndGrant.c b/source/dnode/mnode/impl/src/mndGrant.c index 94acca5f61c2b5650904dd2bf1e84e9e8180ca52..d1d43c841c62cc05d341c8fc20db229d37d91113 100644 --- a/source/dnode/mnode/impl/src/mndGrant.c +++ b/source/dnode/mnode/impl/src/mndGrant.c @@ -14,19 +14,115 @@ */ #define _DEFAULT_SOURCE -#include "os.h" -#include "taoserror.h" #include "mndGrant.h" -#include "mndInt.h" #include "mndShow.h" #ifndef _GRANT -static int32_t mndRetrieveGrant(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) { return TSDB_CODE_OPS_NOT_SUPPORT; } + +static int32_t mndRetrieveGrant(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) { + int32_t numOfRows = 0; + char *pWrite; + int32_t cols = 0; + char tmp[32]; + char tmp1[32]; + + if (pShow->numOfRows < 1) { + cols = 0; + SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + const char *src = "community"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "false"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + cols++; + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + src = "unlimited"; + STR_WITH_SIZE_TO_VARSTR(tmp, src, strlen(src)); + colDataAppend(pColInfo, numOfRows, tmp, false); + + numOfRows++; + } + + pShow->numOfRows += numOfRows; + return numOfRows; +} int32_t mndInitGrant(SMnode *pMnode) { - mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_GRANTS, mndRetrieveGrant); - return TSDB_CODE_SUCCESS; + mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_GRANTS, mndRetrieveGrant); + return 0; } + void mndCleanupGrant() {} void grantParseParameter() { mError("can't parsed parameter k"); } int32_t grantCheck(EGrantType grant) { return TSDB_CODE_SUCCESS; } diff --git a/source/dnode/mnode/impl/src/mndMnode.c b/source/dnode/mnode/impl/src/mndMnode.c index 027de66b42ddf4cf6a42fbf5e02e39e50f479106..c5a4282b3bdfcee8454bb341c9c4c7c99dfceb75 100644 --- a/source/dnode/mnode/impl/src/mndMnode.c +++ b/source/dnode/mnode/impl/src/mndMnode.c @@ -397,17 +397,17 @@ static int32_t mndProcessCreateMnodeReq(SRpcMsg *pReq) { goto _OVER; } - if (sdbGetSize(pMnode->pSdb, SDB_MNODE) >= 3) { - terrno = TSDB_CODE_MND_TOO_MANY_MNODES; - goto _OVER; - } - pDnode = mndAcquireDnode(pMnode, createReq.dnodeId); if (pDnode == NULL) { terrno = TSDB_CODE_MND_DNODE_NOT_EXIST; goto _OVER; } + if (sdbGetSize(pMnode->pSdb, SDB_MNODE) >= 3) { + terrno = TSDB_CODE_MND_TOO_MANY_MNODES; + goto _OVER; + } + if (!mndIsDnodeOnline(pDnode, taosGetTimestampMs())) { terrno = TSDB_CODE_NODE_OFFLINE; goto _OVER; @@ -597,6 +597,11 @@ static int32_t mndProcessDropMnodeReq(SRpcMsg *pReq) { goto _OVER; } + if (!mndIsDnodeOnline(pObj->pDnode, taosGetTimestampMs())) { + terrno = TSDB_CODE_NODE_OFFLINE; + goto _OVER; + } + pUser = mndAcquireUser(pMnode, pReq->conn.user); if (pUser == NULL) { terrno = TSDB_CODE_MND_NO_USER_FROM_CONN; diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c index a4ddf966300aca2974fa90f802bac7ed2aeacacb..2454a4686d68fe00de8e0062a25f4e49bffa815a 100644 --- a/source/dnode/mnode/impl/src/mndScheduler.c +++ b/source/dnode/mnode/impl/src/mndScheduler.c @@ -351,9 +351,15 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) { ASSERT(totLevel <= 2); pStream->tasks = taosArrayInit(totLevel, sizeof(void*)); - bool hasExtraSink = false; - bool externalTargetDB = strcmp(pStream->sourceDb, pStream->targetDb) != 0; - if (totLevel == 2 || externalTargetDB) { + bool hasExtraSink = false; + bool externalTargetDB = strcmp(pStream->sourceDb, pStream->targetDb) != 0; + SDbObj* pDbObj = mndAcquireDb(pMnode, pStream->targetDb); + ASSERT(pDbObj != NULL); + sdbRelease(pSdb, pDbObj); + + bool multiTarget = pDbObj->cfg.numOfVgroups > 1; + + if (totLevel == 2 || externalTargetDB || multiTarget) { SArray* taskOneLevel = taosArrayInit(0, sizeof(void*)); taosArrayPush(pStream->tasks, &taskOneLevel); // add extra sink diff --git a/source/dnode/mnode/sdb/src/sdbFile.c b/source/dnode/mnode/sdb/src/sdbFile.c index 9a0afb51ea0dd90431457bdf6493dbbd58b5fa6d..34f5d6f23d541710cbc475f433d1c8d5ff8b4c3d 100644 --- a/source/dnode/mnode/sdb/src/sdbFile.c +++ b/source/dnode/mnode/sdb/src/sdbFile.c @@ -240,7 +240,7 @@ static int32_t sdbReadFileImp(SSdb *pSdb) { if (pFile == NULL) { taosMemoryFree(pRaw); terrno = TAOS_SYSTEM_ERROR(errno); - mError("failed to read sdb file:%s since %s", file, terrstr()); + mDebug("failed to read sdb file:%s since %s", file, terrstr()); return 0; } diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index e810bb99086cc32a580717fb1633ce92682c0ac5..4a5ea49d79dce40c9bfaafd583546d3a80feb9e4 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -136,6 +136,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { pReq->subKey); return -1; } + if (pHandle->consumerId != consumerId) { tqError("tmq poll: consumer handle mismatch for consumer %ld in vg %d, subkey %s, handle consumer id %ld", consumerId, pTq->pVnode->config.vgId, pReq->subKey, pHandle->consumerId); diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 085824b29df08ebcea467646688cd4efedee2b2a..33661ed4dcd25512e1b25b849a90c9255bc2e11a 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -142,10 +142,8 @@ typedef struct SElapsedInfo { typedef struct SHistoFuncBin { double lower; double upper; - union { - int64_t count; - double percentage; - }; + int64_t count; + double percentage; } SHistoFuncBin; typedef struct SHistoFuncInfo { @@ -3106,7 +3104,7 @@ int32_t spreadCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); SSpreadInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); - spreadTransferInfo(pDBuf, pSBuf); + spreadTransferInfo(pSBuf, pDBuf); pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes); return TSDB_CODE_SUCCESS; } @@ -3277,7 +3275,7 @@ int32_t elapsedCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); SElapsedInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); - elapsedTransferInfo(pDBuf, pSBuf); + elapsedTransferInfo(pSBuf, pDBuf); pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes); return TSDB_CODE_SUCCESS; } @@ -3584,7 +3582,7 @@ int32_t histogramCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); SHistoFuncInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); - histogramTransferInfo(pDBuf, pSBuf); + histogramTransferInfo(pSBuf, pDBuf); pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes); return TSDB_CODE_SUCCESS; } @@ -3780,7 +3778,7 @@ int32_t hllCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx) { SResultRowEntryInfo* pSResInfo = GET_RES_INFO(pSourceCtx); SHLLInfo* pSBuf = GET_ROWCELL_INTERBUF(pSResInfo); - hllTransferInfo(pDBuf, pSBuf); + hllTransferInfo(pSBuf, pDBuf); pDResInfo->numOfRes = TMAX(pDResInfo->numOfRes, pSResInfo->numOfRes); return TSDB_CODE_SUCCESS; } diff --git a/source/libs/parser/src/parAstCreater.c b/source/libs/parser/src/parAstCreater.c index 054912d540dac2bac00011ec5d2caffbb5ca4a6a..e7f4b05bc8bd3168ce8b41a211ed852dc2b3d1f7 100644 --- a/source/libs/parser/src/parAstCreater.c +++ b/source/libs/parser/src/parAstCreater.c @@ -801,7 +801,8 @@ SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOpti ((SDatabaseOptions*)pOptions)->pRetentions = pVal; break; case DB_OPTION_SCHEMALESS: - ((SDatabaseOptions*)pOptions)->schemaless = taosStr2Int8(((SToken*)pVal)->z, NULL, 10); +// ((SDatabaseOptions*)pOptions)->schemaless = taosStr2Int8(((SToken*)pVal)->z, NULL, 10); + ((SDatabaseOptions*)pOptions)->schemaless = 1; break; default: break; diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c index 54c8a1821809ffe867addf788027b54123c1a989..c721491489f298b2f95940bf3b3ef9a3de0cb412 100644 --- a/source/libs/parser/src/parInsert.c +++ b/source/libs/parser/src/parInsert.c @@ -1297,11 +1297,12 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) { } static int32_t checkSchemalessDb(SInsertParseContext* pCxt, char* pDbName) { - SDbCfgInfo pInfo = {0}; - char fullName[TSDB_TABLE_FNAME_LEN]; - snprintf(fullName, sizeof(fullName), "%d.%s", pCxt->pComCxt->acctId, pDbName); - CHECK_CODE(getDBCfg(pCxt, fullName, &pInfo)); - return pInfo.schemaless ? TSDB_CODE_SML_INVALID_DB_CONF : TSDB_CODE_SUCCESS; +// SDbCfgInfo pInfo = {0}; +// char fullName[TSDB_TABLE_FNAME_LEN]; +// snprintf(fullName, sizeof(fullName), "%d.%s", pCxt->pComCxt->acctId, pDbName); +// CHECK_CODE(getDBCfg(pCxt, fullName, &pInfo)); +// return pInfo.schemaless ? TSDB_CODE_SML_INVALID_DB_CONF : TSDB_CODE_SUCCESS; + return TSDB_CODE_SUCCESS; } // tb_name @@ -2119,9 +2120,11 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS isOrdered = false; } if (index < 0) { + uError("smlBoundColumnData. index:%d", index); return TSDB_CODE_SML_INVALID_DATA; } if (pColList->cols[index].valStat == VAL_STAT_HAS) { + uError("smlBoundColumnData. already set. index:%d", index); return TSDB_CODE_SML_INVALID_DATA; } lastColIdx = index; diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 8c61c372fe325e65e2c49eaa168a65f1c9424e46..84bded8a58cf75313dcee0c90ffb1dbe09fa6e2a 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2762,15 +2762,16 @@ static int32_t checkTableSchema(STranslateContext* pCxt, SCreateTableStmt* pStmt } static int32_t checkSchemalessDb(STranslateContext* pCxt, const char* pDbName) { - if (0 != pCxt->pParseCxt->schemalessType) { - return TSDB_CODE_SUCCESS; - } - SDbCfgInfo info = {0}; - int32_t code = getDBCfg(pCxt, pDbName, &info); - if (TSDB_CODE_SUCCESS == code) { - code = info.schemaless ? TSDB_CODE_SML_INVALID_DB_CONF : TSDB_CODE_SUCCESS; - } - return code; +// if (0 != pCxt->pParseCxt->schemalessType) { +// return TSDB_CODE_SUCCESS; +// } +// SDbCfgInfo info = {0}; +// int32_t code = getDBCfg(pCxt, pDbName, &info); +// if (TSDB_CODE_SUCCESS == code) { +// code = info.schemaless ? TSDB_CODE_SML_INVALID_DB_CONF : TSDB_CODE_SUCCESS; +// } +// return code; + return TSDB_CODE_SUCCESS; } static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt) { @@ -5048,6 +5049,9 @@ static int32_t buildModifyVnodeArray(STranslateContext* pCxt, SAlterTableStmt* p static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) { SAlterTableStmt* pStmt = (SAlterTableStmt*)pQuery->pRoot; int32_t code = checkSchemalessDb(pCxt, pStmt->dbName); + if (TSDB_CODE_SUCCESS != code) { + return code; + } STableMeta* pTableMeta = NULL; code = getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pTableMeta); if (TSDB_CODE_SUCCESS != code) { diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index 069595390db7833e025f3f9498e169cfb6980581..d3828d9ee446f31676e4b0fb4f5ec7c125682d66 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -110,9 +110,7 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDisp return 0; } // continue dispatch - if (pTask->dispatchType != TASK_DISPATCH__NONE) { - streamDispatch(pTask, pMsgCb); - } + streamDispatch(pTask, pMsgCb); return 0; } diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index d52337463874ab656c697ff72ce5123f5ac4ed64..59ec2b5cebcce508cc7b75bcea446ca17f9ddc43 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -182,6 +182,7 @@ FAIL: } int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb) { + ASSERT(pTask->dispatchType != TASK_DISPATCH__NONE); #if 1 int8_t old = atomic_val_compare_exchange_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL, TASK_OUTPUT_STATUS__WAIT); diff --git a/source/libs/sync/src/syncAppendEntries.c b/source/libs/sync/src/syncAppendEntries.c index b33f3481e72a8654ff4048828b781b0b2193fe2e..29ee26375692a041d8492259147fbf29e191aac5 100644 --- a/source/libs/sync/src/syncAppendEntries.c +++ b/source/libs/sync/src/syncAppendEntries.c @@ -713,7 +713,7 @@ static int32_t syncNodeMakeLogSame(SSyncNode* ths, SyncAppendEntries* pMsg) { // delete confict entries code = ths->pLogStore->syncLogTruncate(ths->pLogStore, delBegin); ASSERT(code == 0); - sInfo("sync event vgId:%d log truncate, from %ld to %ld", ths->vgId, delBegin, delEnd); + sDebug("vgId:%d sync event log truncate, from %ld to %ld", ths->vgId, delBegin, delEnd); logStoreSimpleLog2("after syncNodeMakeLogSame", ths->pLogStore); return code; @@ -994,7 +994,7 @@ int32_t syncNodeOnAppendEntriesSnapshotCb(SSyncNode* ths, SyncAppendEntries* pMs SyncIndex commitEnd = snapshot.lastApplyIndex; ths->commitIndex = snapshot.lastApplyIndex; - sInfo("sync event vgId:%d commit by snapshot from index:%ld to index:%ld, %s", ths->vgId, commitBegin, + sDebug("vgId:%d sync event commit by snapshot from index:%ld to index:%ld, %s", ths->vgId, commitBegin, commitEnd, syncUtilState2String(ths->state)); } diff --git a/source/libs/sync/src/syncAppendEntriesReply.c b/source/libs/sync/src/syncAppendEntriesReply.c index 5caf814cc5fab48a8447ab7b88f2d879b09cb122..0e116e13ee5436c7fde70ec15c5402cfe0d5f7d4 100644 --- a/source/libs/sync/src/syncAppendEntriesReply.c +++ b/source/libs/sync/src/syncAppendEntriesReply.c @@ -189,16 +189,16 @@ int32_t syncNodeOnAppendEntriesReplySnapshotCb(SSyncNode* ths, SyncAppendEntries if (gRaftDetailLog) { char* s = snapshotSender2Str(pSender); - sInfo( - "sync event vgId:%d snapshot send to %s:%d start sender first time, lastApplyIndex:%ld lastApplyTerm:%lu " + sDebug( + "vgId:%d sync event snapshot send to %s:%d start sender first time, lastApplyIndex:%ld lastApplyTerm:%lu " "lastConfigIndex:%ld" "sender:%s", ths->vgId, host, port, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex, s); taosMemoryFree(s); } else { - sInfo( - "sync event vgId:%d snapshot send to %s:%d start sender first time, lastApplyIndex:%ld " + sDebug( + "vgId:%d sync event snapshot send to %s:%d start sender first time, lastApplyIndex:%ld " "lastApplyTerm:%lu lastConfigIndex:%ld", ths->vgId, host, port, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex); diff --git a/source/libs/sync/src/syncCommit.c b/source/libs/sync/src/syncCommit.c index 8236301f8e33056f7d835ea46d9a362b22365d9e..eac9c1fbd89cb111eeca11bbd449a80dc33d699e 100644 --- a/source/libs/sync/src/syncCommit.c +++ b/source/libs/sync/src/syncCommit.c @@ -56,7 +56,7 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) { SyncIndex commitEnd = snapshot.lastApplyIndex; pSyncNode->commitIndex = snapshot.lastApplyIndex; - sInfo("sync event vgId:%d commit by snapshot from index:%ld to index:%ld, %s", pSyncNode->vgId, + sDebug("vgId:%d sync event commit by snapshot from index:%ld to index:%ld, %s", pSyncNode->vgId, pSyncNode->commitIndex, snapshot.lastApplyIndex, syncUtilState2String(pSyncNode->state)); } diff --git a/source/libs/sync/src/syncMain.c b/source/libs/sync/src/syncMain.c index c480df0ec0ae9819b3a67ccc874c777a1a972477..4d410644bb85da1e1e42ab3e982d8d8eb62d6e7e 100644 --- a/source/libs/sync/src/syncMain.c +++ b/source/libs/sync/src/syncMain.c @@ -470,7 +470,7 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { return TAOS_SYNC_PROPOSE_OTHER_ERROR; } assert(rid == pSyncNode->rid); - sTrace("sync event vgId:%d propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType)); + sDebug("vgId:%d sync event propose msgType:%s", pSyncNode->vgId, TMSG_INFO(pMsg->msgType)); if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { SRespStub stub; @@ -501,7 +501,7 @@ int32_t syncPropose(int64_t rid, const SRpcMsg* pMsg, bool isWeak) { SSyncNode* syncNodeOpen(const SSyncInfo* pOldSyncInfo) { SSyncInfo* pSyncInfo = (SSyncInfo*)pOldSyncInfo; - sInfo("sync event vgId:%d sync open", pSyncInfo->vgId); + sDebug("vgId:%d sync event sync open", pSyncInfo->vgId); SSyncNode* pSyncNode = (SSyncNode*)taosMemoryMalloc(sizeof(SSyncNode)); assert(pSyncNode != NULL); @@ -761,7 +761,7 @@ void syncNodeStartStandBy(SSyncNode* pSyncNode) { } void syncNodeClose(SSyncNode* pSyncNode) { - sInfo("sync event vgId:%d sync close", pSyncNode->vgId); + sDebug("vgId:%d sync event sync close", pSyncNode->vgId); int32_t ret; assert(pSyncNode != NULL); @@ -1240,7 +1240,8 @@ void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term) { } void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) { - sInfo("sync event vgId:%d become follower, isStandBy:%d, %s", pSyncNode->vgId, pSyncNode->pRaftCfg->isStandBy, debugStr); + sDebug("vgId:%d sync event become follower, isStandBy:%d, %s", pSyncNode->vgId, pSyncNode->pRaftCfg->isStandBy, + debugStr); // maybe clear leader cache if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) { @@ -1274,7 +1275,8 @@ void syncNodeBecomeFollower(SSyncNode* pSyncNode, const char* debugStr) { // /\ UNCHANGED <> // void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) { - sInfo("sync event vgId:%d become leader, isStandBy:%d, %s", pSyncNode->vgId, pSyncNode->pRaftCfg->isStandBy, debugStr); + sDebug("vgId:%d sync event become leader, isStandBy:%d, %s", pSyncNode->vgId, pSyncNode->pRaftCfg->isStandBy, + debugStr); // state change pSyncNode->state = TAOS_SYNC_STATE_LEADER; @@ -1882,7 +1884,7 @@ static int32_t syncNodeConfigChange(SSyncNode* ths, SRpcMsg* pRpcMsg, SSyncRaftE int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, uint64_t flag) { int32_t code = 0; ESyncState state = flag; - sInfo("sync event vgId:%d commit by wal from index:%" PRId64 " to index:%" PRId64 ", %s", ths->vgId, beginIndex, + sDebug("vgId:%d sync event commit by wal from index:%" PRId64 " to index:%" PRId64 ", %s", ths->vgId, beginIndex, endIndex, syncUtilState2String(state)); // execute fsm @@ -1931,7 +1933,7 @@ int32_t syncNodeCommit(SSyncNode* ths, SyncIndex beginIndex, SyncIndex endIndex, ths->pFsm->FpRestoreFinishCb(ths->pFsm); } ths->restoreFinish = true; - sInfo("sync event vgId:%d restore finish", ths->vgId); + sDebug("vgId:%d sync event restore finish", ths->vgId); } } diff --git a/source/libs/sync/src/syncRaftLog.c b/source/libs/sync/src/syncRaftLog.c index 92699ab24defee238a7ccf42397d29dca5a0d950..996cd12e4a6174ab6c23299657b68783983ba015 100644 --- a/source/libs/sync/src/syncRaftLog.c +++ b/source/libs/sync/src/syncRaftLog.c @@ -162,7 +162,7 @@ static int32_t raftLogAppendEntry(struct SSyncLogStore* pLogStore, SSyncRaftEntr walFsync(pWal, true); - sTrace("sync event vgId:%d write index:%ld, %s, isStandBy:%d, msgType:%s, originalRpcType:%s", pData->pSyncNode->vgId, + sDebug("vgId:%d sync event write index:%ld, %s, isStandBy:%d, msgType:%s, originalRpcType:%s", pData->pSyncNode->vgId, pEntry->index, syncUtilState2String(pData->pSyncNode->state), pData->pSyncNode->pRaftCfg->isStandBy, TMSG_INFO(pEntry->msgType), TMSG_INFO(pEntry->originalRpcType)); @@ -320,7 +320,7 @@ int32_t logStoreAppendEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry) { walFsync(pWal, true); - sTrace("sync event old write wal: %ld", pEntry->index); + sDebug("sync event old write wal: %ld", pEntry->index); return code; } diff --git a/source/libs/sync/src/syncSnapshot.c b/source/libs/sync/src/syncSnapshot.c index 36598cc2bd1aa6625d5fb550fb78127d6a98095d..38742220feb88ca12b712933969ee2aa3ee3468e 100644 --- a/source/libs/sync/src/syncSnapshot.c +++ b/source/libs/sync/src/syncSnapshot.c @@ -140,16 +140,16 @@ void snapshotSenderStart(SSyncSnapshotSender *pSender) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace( - "sync event vgId:%d snapshot send to %s:%d begin seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " + sDebug( + "vgId:%d sync event snapshot send to %s:%d begin seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " "lastConfigIndex:%ld send " "msg:%s", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex, msgStr); taosMemoryFree(msgStr); } else { - sTrace( - "sync event vgId:%d snapshot send to %s:%d begin seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " + sDebug( + "vgId:%d sync event snapshot send to %s:%d begin seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " "lastConfigIndex:%ld", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex); @@ -278,23 +278,23 @@ int32_t snapshotSend(SSyncSnapshotSender *pSender) { if (pSender->seq == SYNC_SNAPSHOT_SEQ_END) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace( - "sync event vgId:%d snapshot send to %s:%d finish seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " + sDebug( + "vgId:%d sync event snapshot send to %s:%d finish seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " "lastConfigIndex:%ld send " "msg:%s", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex, msgStr); taosMemoryFree(msgStr); } else { - sTrace( - "sync event vgId:%d snapshot send to %s:%d finish seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " + sDebug( + "vgId:%d sync event snapshot send to %s:%d finish seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " "lastConfigIndex:%ld", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex); } } else { - sTrace( - "sync event vgId:%d snapshot send to %s:%d sending seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " + sDebug( + "vgId:%d sync event snapshot send to %s:%d sending seq:%d ack:%d lastApplyIndex:%ld lastApplyTerm:%lu " "lastConfigIndex:%ld", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack, pSender->snapshot.lastApplyIndex, pSender->snapshot.lastApplyTerm, pSender->snapshot.lastConfigIndex); @@ -328,11 +328,11 @@ int32_t snapshotReSend(SSyncSnapshotSender *pSender) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace("sync event vgId:%d snapshot send to %s:%d resend seq:%d ack:%d send msg:%s", pSender->pSyncNode->vgId, + sDebug("vgId:%d sync event snapshot send to %s:%d resend seq:%d ack:%d send msg:%s", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack, msgStr); taosMemoryFree(msgStr); } else { - sTrace("sync event vgId:%d snapshot send to %s:%d resend seq:%d ack:%d", pSender->pSyncNode->vgId, host, port, + sDebug("vgId:%d sync event snapshot send to %s:%d resend seq:%d ack:%d", pSender->pSyncNode->vgId, host, port, pSender->seq, pSender->ack); } @@ -565,12 +565,17 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace("sync event vgId:%d snapshot recv from %s:%d begin ack:%d, lastIndex:%ld, lastTerm:%lu, recv msg:%s", - pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, msgStr); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d begin ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld, recv msg:%s", + pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex, + msgStr); taosMemoryFree(msgStr); } else { - sTrace("sync event vgId:%d snapshot recv from %s:%d begin ack:%d, lastIndex:%ld, lastTerm:%lu", - pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d begin ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld", + pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex); } } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_END) { @@ -597,12 +602,12 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { bool isDrop; if (IamInNew) { - sTrace("sync event vgId:%d update config by snapshot, lastIndex:%ld, lastTerm:%lu, lastConfigIndex:%ld ", + sDebug("vgId:%d sync event update config by snapshot, lastIndex:%ld, lastTerm:%lu, lastConfigIndex:%ld ", pSyncNode->vgId, pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex); syncNodeUpdateConfig(pSyncNode, &newSyncCfg, pMsg->lastConfigIndex, &isDrop); } else { - sTrace( - "sync event vgId:%d do not update config by snapshot, I am not in newCfg, lastIndex:%ld, lastTerm:%lu, " + sDebug( + "vgId:%d sync event do not update config by snapshot, I am not in newCfg, lastIndex:%ld, lastTerm:%lu, " "lastConfigIndex:%ld ", pSyncNode->vgId, pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex); } @@ -626,19 +631,20 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { if (gRaftDetailLog) { char *logSimpleStr = logStoreSimple2Str(pSyncNode->pLogStore); - sInfo( - "sync event vgId:%d snapshot recv from %s:%d finish, update log begin index:%ld, " + sDebug( + "vgId:%d sync event snapshot recv from %s:%d finish, update log begin index:%ld, " "snapshot.lastApplyIndex:%ld, " - "snapshot.lastApplyTerm:%lu, raft log:%s", + "snapshot.lastApplyTerm:%lu, snapshot.lastConfigIndex:%ld, raft log:%s", pSyncNode->vgId, host, port, pMsg->lastIndex + 1, snapshot.lastApplyIndex, snapshot.lastApplyTerm, - logSimpleStr); + snapshot.lastConfigIndex, logSimpleStr); taosMemoryFree(logSimpleStr); } else { - sInfo( - "sync event vgId:%d snapshot recv from %s:%d finish, update log begin index:%ld, " + sDebug( + "vgId:%d sync event snapshot recv from %s:%d finish, update log begin index:%ld, " "snapshot.lastApplyIndex:%ld, " - "snapshot.lastApplyTerm:%lu", - pSyncNode->vgId, host, port, pMsg->lastIndex + 1, snapshot.lastApplyIndex, snapshot.lastApplyTerm); + "snapshot.lastApplyTerm:%lu, snapshot.lastConfigIndex:%ld", + pSyncNode->vgId, host, port, pMsg->lastIndex + 1, snapshot.lastApplyIndex, snapshot.lastApplyTerm, + snapshot.lastConfigIndex); } pReceiver->pWriter = NULL; @@ -648,12 +654,18 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace("sync event vgId:%d snapshot recv from %s:%d end ack:%d, lastIndex:%ld, lastTerm:%lu, recv msg:%s", - pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, msgStr); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d end ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld, recv msg:%s", + pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, + pMsg->lastConfigIndex, msgStr); taosMemoryFree(msgStr); } else { - sTrace("sync event vgId:%d snapshot recv from %s:%d end ack:%d, lastIndex:%ld, lastTerm:%lu", - pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d end ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld", + pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, + pMsg->lastConfigIndex); } } else if (pMsg->seq == SYNC_SNAPSHOT_SEQ_FORCE_CLOSE) { @@ -667,14 +679,19 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace( - "sync event vgId:%d snapshot recv from %s:%d force close ack:%d, lastIndex:%ld, lastTerm:%lu, recv " + sDebug( + "vgId:%d sync event snapshot recv from %s:%d force close ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld, recv " "msg:%s", - pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, msgStr); + pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, + pMsg->lastConfigIndex, msgStr); taosMemoryFree(msgStr); } else { - sTrace("sync event vgId:%d snapshot recv from %s:%d force close ack:%d, lastIndex:%ld, lastTerm:%lu", - pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d force close ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld", + pReceiver->pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, + pMsg->lastConfigIndex); } } else if (pMsg->seq > SYNC_SNAPSHOT_SEQ_BEGIN && pMsg->seq < SYNC_SNAPSHOT_SEQ_END) { @@ -693,13 +710,17 @@ int32_t syncNodeOnSnapshotSendCb(SSyncNode *pSyncNode, SyncSnapshotSend *pMsg) { if (gRaftDetailLog) { char *msgStr = syncSnapshotSend2Str(pMsg); - sTrace( - "sync event vgId:%d snapshot recv from %s:%d receiving ack:%d, lastIndex:%ld, lastTerm:%lu, recv msg:%s", - pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, msgStr); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d receiving ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld, recv msg:%s", + pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex, + msgStr); taosMemoryFree(msgStr); } else { - sTrace("sync event vgId:%d snapshot recv from %s:%d receiving ack:%d, lastIndex:%ld, lastTerm:%lu", - pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm); + sDebug( + "vgId:%d sync event snapshot recv from %s:%d receiving ack:%d, lastIndex:%ld, lastTerm:%lu, " + "lastConfigIndex:%ld", + pSyncNode->vgId, host, port, pReceiver->ack, pMsg->lastIndex, pMsg->lastTerm, pMsg->lastConfigIndex); } } else { diff --git a/source/util/src/tlrucache.c b/source/util/src/tlrucache.c new file mode 100644 index 0000000000000000000000000000000000000000..b034a6e73e9794926fb606fd0b3c2908a58ca7a7 --- /dev/null +++ b/source/util/src/tlrucache.c @@ -0,0 +1,794 @@ +/* + * 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 . + */ + +#define _DEFAULT_SOURCE +#include "tlrucache.h" +#include "os.h" +#include "tdef.h" +#include "taoserror.h" +#include "tlog.h" +#include "tarray.h" + +typedef struct SLRUEntry SLRUEntry; +typedef struct SLRUEntryTable SLRUEntryTable; +typedef struct SLRUCacheShard SLRUCacheShard; +typedef struct SShardedCache SShardedCache; + +enum { + TAOS_LRU_IN_CACHE = (1 << 0), // Whether this entry is referenced by the hash table. + + TAOS_LRU_IS_HIGH_PRI = (1 << 1), // Whether this entry is high priority entry. + + TAOS_LRU_IN_HIGH_PRI_POOL = (1 << 2), // Whether this entry is in high-pri pool. + + TAOS_LRU_HAS_HIT = (1 << 3), // Whether this entry has had any lookups (hits). +}; + +struct SLRUEntry { + void *value; + _taos_lru_deleter_t deleter; + SLRUEntry *nextHash; + SLRUEntry *next; + SLRUEntry *prev; + size_t totalCharge; + size_t keyLength; + uint32_t hash; + uint32_t refs; + uint8_t flags; + char keyData[1]; +}; + +#define TAOS_LRU_ENTRY_IN_CACHE(h) ((h)->flags & TAOS_LRU_IN_CACHE) +#define TAOS_LRU_ENTRY_IN_HIGH_POOL(h) ((h)->flags & TAOS_LRU_IN_HIGH_PRI_POOL) +#define TAOS_LRU_ENTRY_IS_HIGH_PRI(h) ((h)->flags & TAOS_LRU_IS_HIGH_PRI) +#define TAOS_LRU_ENTRY_HAS_HIT(h) ((h)->flags & TAOS_LRU_HAS_HIT) + +#define TAOS_LRU_ENTRY_SET_IN_CACHE(h, inCache) do { if(inCache) {(h)->flags |= TAOS_LRU_IN_CACHE;} else {(h)->flags &= ~TAOS_LRU_IN_CACHE;} } while(0) +#define TAOS_LRU_ENTRY_SET_IN_HIGH_POOL(h, inHigh) do { if(inHigh) {(h)->flags |= TAOS_LRU_IN_HIGH_PRI_POOL;} else {(h)->flags &= ~TAOS_LRU_IN_HIGH_PRI_POOL;} } while(0) +#define TAOS_LRU_ENTRY_SET_PRIORITY(h, priority) do { if(priority == TAOS_LRU_PRIORITY_HIGH) {(h)->flags |= TAOS_LRU_IS_HIGH_PRI;} else {(h)->flags &= ~TAOS_LRU_IS_HIGH_PRI;} } while(0) +#define TAOS_LRU_ENTRY_SET_HIT(h) ((h)->flags |= TAOS_LRU_HAS_HIT) + +#define TAOS_LRU_ENTRY_HAS_REFS(h) ((h)->refs > 0) +#define TAOS_LRU_ENTRY_REF(h) (++(h)->refs) + +static bool taosLRUEntryUnref(SLRUEntry *entry) { + assert(entry->refs > 0); + --entry->refs; + return entry->refs == 0; +} + +static void taosLRUEntryFree(SLRUEntry *entry) { + assert(entry->refs == 0); + + if (entry->deleter) { + (*entry->deleter)(entry->keyData, entry->keyLength, entry->value); + } + + taosMemoryFree(entry); +} + +typedef void (*_taos_lru_table_func_t)(SLRUEntry *entry); + +struct SLRUEntryTable { + int lengthBits; + SLRUEntry **list; + uint32_t elems; + int maxLengthBits; +}; + +static int taosLRUEntryTableInit(SLRUEntryTable *table, int maxUpperHashBits) { + table->lengthBits = 4; + table->list = taosMemoryCalloc(1 << table->lengthBits, sizeof(SLRUEntry*)); + if (!table->list) { + return -1; + } + + table->elems = 0; + table->maxLengthBits = maxUpperHashBits; + + return 0; +} + +static void taosLRUEntryTableApply(SLRUEntryTable *table, _taos_lru_table_func_t func, uint32_t begin, uint32_t end) { + for (uint32_t i = begin; i < end; ++i) { + SLRUEntry *h = table->list[i]; + while (h) { + SLRUEntry *n = h->nextHash; + assert(TAOS_LRU_ENTRY_IN_CACHE(h)); + func(h); + h = n; + } + } +} + +static void taosLRUEntryTableFree(SLRUEntry *entry) { + if (!TAOS_LRU_ENTRY_HAS_REFS(entry)) { + taosLRUEntryFree(entry); + } +} + +static void taosLRUEntryTableCleanup(SLRUEntryTable *table) { + taosLRUEntryTableApply(table, taosLRUEntryTableFree, 0, 1 << table->lengthBits); + + taosMemoryFree(table->list); +} + +static SLRUEntry **taosLRUEntryTableFindPtr(SLRUEntryTable * table, const void *key, size_t keyLen, uint32_t hash) { + SLRUEntry **entry = &table->list[hash >> (32 - table->lengthBits)]; + while (*entry && ((*entry)->hash != hash || memcmp(key, (*entry)->keyData, keyLen) != 0)) { + entry = &(*entry)->nextHash; + } + + return entry; +} + +static void taosLRUEntryTableResize(SLRUEntryTable * table) { + int lengthBits = table->lengthBits; + if (lengthBits >= table->maxLengthBits) { + return; + } + + if (lengthBits >= 31) { + return; + } + + uint32_t oldLength = 1 << lengthBits; + int newLengthBits = lengthBits + 1; + SLRUEntry **newList = taosMemoryCalloc(1 << newLengthBits, sizeof(SLRUEntry*)); + if (!newList) { + return; + } + uint32_t count = 0; + for (uint32_t i = 0; i < oldLength; ++i) { + SLRUEntry *entry = table->list[i]; + while (entry) { + SLRUEntry *next = entry->nextHash; + uint32_t hash = entry->hash; + SLRUEntry **ptr = &newList[hash >> (32 - newLengthBits)]; + entry->nextHash = *ptr; + *ptr = entry; + entry = next; + ++count; + } + } + assert(table->elems == count); + + taosMemoryFree(table->list); + table->list = newList; + table->lengthBits = newLengthBits; +} + +static SLRUEntry *taosLRUEntryTableLookup(SLRUEntryTable * table, const void *key, size_t keyLen, uint32_t hash) { + return *taosLRUEntryTableFindPtr(table, key, keyLen, hash); +} + +static SLRUEntry *taosLRUEntryTableInsert(SLRUEntryTable * table, SLRUEntry *entry) { + SLRUEntry **ptr = taosLRUEntryTableFindPtr(table, entry->keyData, entry->keyLength, entry->hash); + SLRUEntry *old = *ptr; + entry->nextHash = (old == NULL) ? NULL : old->nextHash; + *ptr = entry; + if (old == NULL) { + ++table->elems; + if ((table->elems >> table->lengthBits) > 0) { + taosLRUEntryTableResize(table); + } + } + + return old; +} + +static SLRUEntry *taosLRUEntryTableRemove(SLRUEntryTable * table, const void *key, size_t keyLen, uint32_t hash) { + SLRUEntry **entry = taosLRUEntryTableFindPtr(table, key, keyLen, hash); + SLRUEntry *result = *entry; + if (result) { + *entry = result->nextHash; + --table->elems; + } + + return result; +} + +struct SLRUCacheShard { + size_t capacity; + size_t highPriPoolUsage; + bool strictCapacity; + double highPriPoolRatio; + double highPriPoolCapacity; + SLRUEntry lru; + SLRUEntry *lruLowPri; + SLRUEntryTable table; + size_t usage; // Memory size for entries residing in the cache. + size_t lruUsage; // Memory size for entries residing only in the LRU list. + TdThreadMutex mutex; +}; + +#define TAOS_LRU_CACHE_SHARD_HASH32(key, len) (MurmurHash3_32((key), (len))) + +static void taosLRUCacheShardMaintainPoolSize(SLRUCacheShard *shard) { + while (shard->highPriPoolUsage > shard->highPriPoolCapacity) { + shard->lruLowPri = shard->lruLowPri->next; + assert(shard->lruLowPri != &shard->lru); + TAOS_LRU_ENTRY_SET_IN_HIGH_POOL(shard->lruLowPri, false); + + assert(shard->highPriPoolUsage >= shard->lruLowPri->totalCharge); + shard->highPriPoolUsage -= shard->lruLowPri->totalCharge; + } +} + +static void taosLRUCacheShardLRUInsert(SLRUCacheShard *shard, SLRUEntry *e) { + assert(e->next == NULL); + assert(e->prev == NULL); + + if (shard->highPriPoolRatio > 0 + && (TAOS_LRU_ENTRY_IS_HIGH_PRI(e) || TAOS_LRU_ENTRY_HAS_HIT(e))) { + e->next = &shard->lru; + e->prev = shard->lru.prev; + + e->prev->next = e; + e->next->prev = e; + + TAOS_LRU_ENTRY_SET_IN_HIGH_POOL(e, true); + shard->highPriPoolUsage += e->totalCharge; + taosLRUCacheShardMaintainPoolSize(shard); + } else { + e->next = shard->lruLowPri->next; + e->prev = shard->lruLowPri; + + e->prev->next = e; + e->next->prev = e; + + TAOS_LRU_ENTRY_SET_IN_HIGH_POOL(e, false); + shard->lruLowPri = e; + } + + shard->lruUsage += e->totalCharge; +} + +static void taosLRUCacheShardLRURemove(SLRUCacheShard *shard, SLRUEntry *e) { + assert(e->next); + assert(e->prev); + + if (shard->lruLowPri == e) { + shard->lruLowPri = e->prev; + } + e->next->prev = e->prev; + e->prev->next = e->next; + e->prev = e->next = NULL; + + assert(shard->lruUsage >= e->totalCharge); + shard->lruUsage -= e->totalCharge; + if (TAOS_LRU_ENTRY_IN_HIGH_POOL(e)) { + assert(shard->highPriPoolUsage >= e->totalCharge); + shard->highPriPoolUsage -= e->totalCharge; + } +} + +static void taosLRUCacheShardEvictLRU(SLRUCacheShard *shard, size_t charge, SArray *deleted) { + while (shard->usage + charge > shard->capacity && shard->lru.next != &shard->lru) { + SLRUEntry *old = shard->lru.next; + assert(TAOS_LRU_ENTRY_IN_CACHE(old) && !TAOS_LRU_ENTRY_HAS_REFS(old)); + + taosLRUCacheShardLRURemove(shard, old); + taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash); + + TAOS_LRU_ENTRY_SET_IN_CACHE(old, false); + assert(shard->usage >= old->totalCharge); + shard->usage -= old->totalCharge; + + taosArrayPush(deleted, &old); + } +} + +static void taosLRUCacheShardSetCapacity(SLRUCacheShard *shard, size_t capacity) { + SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); + + taosThreadMutexLock(&shard->mutex); + + shard->capacity = capacity; + shard->highPriPoolCapacity = capacity * shard->highPriPoolRatio; + taosLRUCacheShardEvictLRU(shard, 0, lastReferenceList); + + taosThreadMutexUnlock(&shard->mutex); + + for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { + SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); + taosLRUEntryFree(entry); + } + taosArrayDestroy(lastReferenceList); +} + +static int taosLRUCacheShardInit(SLRUCacheShard *shard, size_t capacity, bool strict, + double highPriPoolRatio, int maxUpperHashBits) { + if (taosLRUEntryTableInit(&shard->table, maxUpperHashBits) < 0) { + return -1; + } + + taosThreadMutexInit(&shard->mutex, NULL); + + shard->capacity = 0; + shard->highPriPoolUsage = 0; + shard->strictCapacity = strict; + shard->highPriPoolRatio = highPriPoolRatio; + shard->highPriPoolCapacity = 0; + + shard->usage = 0; + shard->lruUsage = 0; + + shard->lru.next = &shard->lru; + shard->lru.prev = &shard->lru; + shard->lruLowPri = &shard->lru; + + taosLRUCacheShardSetCapacity(shard, capacity); + + return 0; +} + +static void taosLRUCacheShardCleanup(SLRUCacheShard *shard) { + taosThreadMutexDestroy(&shard->mutex); + + taosLRUEntryTableCleanup(&shard->table); +} + +static LRUStatus taosLRUCacheShardInsertEntry(SLRUCacheShard *shard, SLRUEntry *e, LRUHandle **handle, bool freeOnFail) { + LRUStatus status = TAOS_LRU_STATUS_OK; + SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); + + taosThreadMutexLock(&shard->mutex); + + taosLRUCacheShardEvictLRU(shard, e->totalCharge, lastReferenceList); + + if (shard->usage + e->totalCharge > shard->capacity && (shard->strictCapacity || handle == NULL)) { + TAOS_LRU_ENTRY_SET_IN_CACHE(e, false); + if (handle == NULL) { + taosArrayPush(lastReferenceList, &e); + } else { + if (freeOnFail) { + taosMemoryFree(e); + + *handle = NULL; + } + + status = TAOS_LRU_STATUS_INCOMPLETE; + } + } else { + SLRUEntry *old = taosLRUEntryTableInsert(&shard->table, e); + shard->usage += e->totalCharge; + if (old != NULL) { + status = TAOS_LRU_STATUS_OK_OVERWRITTEN; + + assert(TAOS_LRU_ENTRY_IN_CACHE(old)); + TAOS_LRU_ENTRY_SET_IN_CACHE(old, false); + if (!TAOS_LRU_ENTRY_HAS_REFS(e)) { + taosLRUCacheShardLRURemove(shard, old); + assert(shard->usage >= old->totalCharge); + shard->usage -= old->totalCharge; + + taosArrayPush(lastReferenceList, &old); + } + } + if (handle == NULL) { + taosLRUCacheShardLRUInsert(shard, e); + } else { + if (!TAOS_LRU_ENTRY_HAS_REFS(e)) { + TAOS_LRU_ENTRY_REF(e); + } + + *handle = (LRUHandle*) e; + } + } + + taosThreadMutexUnlock(&shard->mutex); + + for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { + SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); + + taosLRUEntryFree(entry); + } + taosArrayDestroy(lastReferenceList); + + return status; +} + +static LRUStatus taosLRUCacheShardInsert(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash, + void *value, size_t charge, _taos_lru_deleter_t deleter, + LRUHandle **handle, LRUPriority priority) { + SLRUEntry *e = taosMemoryCalloc(1, sizeof(SLRUEntry) - 1 + keyLen); + if (!e) { + return TAOS_LRU_STATUS_FAIL; + } + + e->value = value; + e->flags = 0; + e->deleter = deleter; + e->keyLength = keyLen; + e->hash = hash; + e->refs = 0; + e->next = e->prev = NULL; + TAOS_LRU_ENTRY_SET_IN_CACHE(e, true); + + TAOS_LRU_ENTRY_SET_PRIORITY(e, priority); + memcpy(e->keyData, key, keyLen); + // TODO: e->CalcTotalCharge(charge, metadataChargePolicy); + e->totalCharge = charge; + + return taosLRUCacheShardInsertEntry(shard, e, handle, true); +} + +static LRUHandle *taosLRUCacheShardLookup(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash) { + SLRUEntry *e = NULL; + + taosThreadMutexLock(&shard->mutex); + e = taosLRUEntryTableLookup(&shard->table, key, keyLen, hash); + if (e != NULL) { + assert(TAOS_LRU_ENTRY_IN_CACHE(e)); + if (!TAOS_LRU_ENTRY_HAS_REFS(e)) { + taosLRUCacheShardLRURemove(shard, e); + } + TAOS_LRU_ENTRY_REF(e); + TAOS_LRU_ENTRY_SET_HIT(e); + } + + taosThreadMutexUnlock(&shard->mutex); + + return (LRUHandle *) e; +} + +static void taosLRUCacheShardErase(SLRUCacheShard *shard, const void *key, size_t keyLen, uint32_t hash) { + bool lastReference = false; + taosThreadMutexLock(&shard->mutex); + + SLRUEntry *e = taosLRUEntryTableRemove(&shard->table, key, keyLen, hash); + if (e != NULL) { + assert(TAOS_LRU_ENTRY_IN_CACHE(e)); + TAOS_LRU_ENTRY_SET_IN_CACHE(e, false); + if (!TAOS_LRU_ENTRY_HAS_REFS(e)) { + taosLRUCacheShardLRURemove(shard, e); + + assert(shard->usage >= e->totalCharge); + shard->usage -= e->totalCharge; + lastReference = true; + } + } + + taosThreadMutexUnlock(&shard->mutex); + + if (lastReference) { + taosLRUEntryFree(e); + } +} + +static void taosLRUCacheShardEraseUnrefEntries(SLRUCacheShard *shard) { + SArray *lastReferenceList = taosArrayInit(16, POINTER_BYTES); + + taosThreadMutexLock(&shard->mutex); + + while (shard->lru.next != &shard->lru) { + SLRUEntry *old = shard->lru.next; + assert(TAOS_LRU_ENTRY_IN_CACHE(old) && !TAOS_LRU_ENTRY_HAS_REFS(old)); + taosLRUCacheShardLRURemove(shard, old); + taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash); + TAOS_LRU_ENTRY_SET_IN_CACHE(old, false); + assert(shard->usage >= old->totalCharge); + shard->usage -= old->totalCharge; + + taosArrayPush(lastReferenceList, &old); + } + + taosThreadMutexUnlock(&shard->mutex); + + for (int i = 0; i < taosArrayGetSize(lastReferenceList); ++i) { + SLRUEntry *entry = taosArrayGetP(lastReferenceList, i); + + taosLRUEntryFree(entry); + } + + taosArrayDestroy(lastReferenceList); +} + +static bool taosLRUCacheShardRef(SLRUCacheShard *shard, LRUHandle *handle) { + SLRUEntry *e = (SLRUEntry *) handle; + taosThreadMutexLock(&shard->mutex); + + assert(TAOS_LRU_ENTRY_HAS_REFS(e)); + TAOS_LRU_ENTRY_REF(e); + + taosThreadMutexUnlock(&shard->mutex); + + return true; +} + +static bool taosLRUCacheShardRelease(SLRUCacheShard *shard, LRUHandle *handle, bool eraseIfLastRef) { + if (handle == NULL) { + return false; + } + + SLRUEntry *e = (SLRUEntry *) handle; + bool lastReference = false; + + taosThreadMutexLock(&shard->mutex); + + lastReference = taosLRUEntryUnref(e); + if (lastReference && TAOS_LRU_ENTRY_IN_CACHE(e)) { + if (shard->usage > shard->capacity || eraseIfLastRef) { + assert(shard->lru.next == &shard->lru || eraseIfLastRef); + + taosLRUEntryTableRemove(&shard->table, e->keyData, e->keyLength, e->hash); + TAOS_LRU_ENTRY_SET_IN_CACHE(e, false); + } else { + taosLRUCacheShardLRUInsert(shard, e); + + lastReference = false; + } + } + + if (lastReference && e->value) { + assert(shard->usage >= e->totalCharge); + shard->usage -= e->totalCharge; + } + + taosThreadMutexUnlock(&shard->mutex); + + if (lastReference) { + taosLRUEntryFree(e); + } + + return lastReference; +} + +static size_t taosLRUCacheShardGetUsage(SLRUCacheShard *shard) { + size_t usage = 0; + + taosThreadMutexLock(&shard->mutex); + usage = shard->usage; + taosThreadMutexUnlock(&shard->mutex); + + return usage; +} + +static size_t taosLRUCacheShardGetPinnedUsage(SLRUCacheShard *shard) { + size_t usage = 0; + + taosThreadMutexLock(&shard->mutex); + + assert(shard->usage >= shard->lruUsage); + usage = shard->usage - shard->lruUsage; + + taosThreadMutexUnlock(&shard->mutex); + + return usage; +} + +static void taosLRUCacheShardSetStrictCapacity(SLRUCacheShard *shard, bool strict) { + taosThreadMutexLock(&shard->mutex); + + shard->strictCapacity = strict; + + taosThreadMutexUnlock(&shard->mutex); +} + +struct SShardedCache { + uint32_t shardMask; + TdThreadMutex capacityMutex; + size_t capacity; + bool strictCapacity; + uint64_t lastId; // atomic var for last id +}; + +struct SLRUCache { + SShardedCache shardedCache; + SLRUCacheShard *shards; + int numShards; +}; + +static int getDefaultCacheShardBits(size_t capacity) { + int numShardBits = 0; + size_t minShardSize = 512 * 1024; + size_t numShards = capacity / minShardSize; + while (numShards >>= 1) { + if (++numShardBits >= 6) { + return numShardBits; + } + } + + return numShardBits; +} + +SLRUCache *taosLRUCacheInit(size_t capacity, int numShardBits, double highPriPoolRatio) { + if (numShardBits >= 20) { + return NULL; + } + if (highPriPoolRatio < 0.0 || highPriPoolRatio > 1.0) { + return NULL; + } + SLRUCache *cache = taosMemoryCalloc(1, sizeof(SLRUCache)); + if (!cache) { + return NULL; + } + + if (numShardBits < 0) { + numShardBits = getDefaultCacheShardBits(capacity); + } + + int numShards = 1 << numShardBits; + cache->shards = taosMemoryCalloc(numShards, sizeof(SLRUCacheShard)); + if (!cache->shards) { + taosMemoryFree(cache); + + return NULL; + } + + bool strictCapacity = 1; + size_t perShard = (capacity + (numShards - 1)) / numShards; + for (int i = 0; i < numShards; ++i) { + taosLRUCacheShardInit(&cache->shards[i], perShard, strictCapacity, highPriPoolRatio, 32 - numShardBits); + } + + cache->numShards = numShards; + + cache->shardedCache.shardMask = (1 << numShardBits) - 1; + cache->shardedCache.strictCapacity = strictCapacity; + cache->shardedCache.capacity = capacity; + cache->shardedCache.lastId = 1; + + taosThreadMutexInit(&cache->shardedCache.capacityMutex, NULL); + + return cache; +} + +void taosLRUCacheCleanup(SLRUCache *cache) { + if (cache) { + if (cache->shards) { + int numShards = cache->numShards; + assert(numShards > 0); + for (int i = 0; i < numShards; ++i) { + taosLRUCacheShardCleanup(&cache->shards[i]); + } + taosMemoryFree(cache->shards); + cache->shards = 0; + } + + taosThreadMutexDestroy(&cache->shardedCache.capacityMutex); + + taosMemoryFree(cache); + } +} + +LRUStatus taosLRUCacheInsert(SLRUCache *cache, const void *key, size_t keyLen, void *value, size_t charge, + _taos_lru_deleter_t deleter, LRUHandle **handle, LRUPriority priority) { + uint32_t hash = TAOS_LRU_CACHE_SHARD_HASH32(key, keyLen); + uint32_t shardIndex = hash & cache->shardedCache.shardMask; + + return taosLRUCacheShardInsert(&cache->shards[shardIndex], key, keyLen, hash, value, charge, deleter, handle, priority); +} + +LRUHandle *taosLRUCacheLookup(SLRUCache *cache, const void *key, size_t keyLen) { + uint32_t hash = TAOS_LRU_CACHE_SHARD_HASH32(key, keyLen); + uint32_t shardIndex = hash & cache->shardedCache.shardMask; + + return taosLRUCacheShardLookup(&cache->shards[shardIndex], key, keyLen, hash); +} + +void taosLRUCacheErase(SLRUCache *cache, const void *key, size_t keyLen) { + uint32_t hash = TAOS_LRU_CACHE_SHARD_HASH32(key, keyLen); + uint32_t shardIndex = hash & cache->shardedCache.shardMask; + + return taosLRUCacheShardErase(&cache->shards[shardIndex], key, keyLen, hash); +} + +void taosLRUCacheEraseUnrefEntries(SLRUCache *cache) { + int numShards = cache->numShards; + for (int i = 0; i < numShards; ++i) { + taosLRUCacheShardEraseUnrefEntries(&cache->shards[i]); + } +} + +bool taosLRUCacheRef(SLRUCache *cache, LRUHandle *handle) { + if (handle == NULL) { + return false; + } + + uint32_t hash = ((SLRUEntry *) handle)->hash; + uint32_t shardIndex = hash & cache->shardedCache.shardMask; + + return taosLRUCacheShardRef(&cache->shards[shardIndex], handle); +} + +bool taosLRUCacheRelease(SLRUCache *cache, LRUHandle *handle, bool eraseIfLastRef) { + if (handle == NULL) { + return false; + } + + uint32_t hash = ((SLRUEntry *) handle)->hash; + uint32_t shardIndex = hash & cache->shardedCache.shardMask; + + return taosLRUCacheShardRelease(&cache->shards[shardIndex], handle, eraseIfLastRef); +} + +void* taosLRUCacheValue(SLRUCache *cache, LRUHandle *handle) { + return ((SLRUEntry*) handle)->value; +} + +size_t taosLRUCacheGetUsage(SLRUCache *cache) { + size_t usage = 0; + + for (int i = 0; i < cache->numShards; ++i) { + usage += taosLRUCacheShardGetUsage(&cache->shards[i]); + } + + return usage; +} + +size_t taosLRUCacheGetPinnedUsage(SLRUCache *cache) { + size_t usage = 0; + + for (int i = 0; i < cache->numShards; ++i) { + usage += taosLRUCacheShardGetPinnedUsage(&cache->shards[i]); + } + + return usage; +} + +void taosLRUCacheSetCapacity(SLRUCache *cache, size_t capacity) { + uint32_t numShards = cache->numShards; + size_t perShard = (capacity + (numShards = 1)) / numShards; + + taosThreadMutexLock(&cache->shardedCache.capacityMutex); + + for (int i = 0; i < numShards; ++i) { + taosLRUCacheShardSetCapacity(&cache->shards[i], perShard); + } + + cache->shardedCache.capacity = capacity; + + taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); +} + +size_t taosLRUCacheGetCapacity(SLRUCache *cache) { + size_t capacity = 0; + + taosThreadMutexLock(&cache->shardedCache.capacityMutex); + + capacity = cache->shardedCache.capacity; + + taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); + + return capacity; +} + +void taosLRUCacheSetStrictCapacity(SLRUCache *cache, bool strict) { + uint32_t numShards = cache->numShards; + + taosThreadMutexLock(&cache->shardedCache.capacityMutex); + + for (int i = 0; i < numShards; ++i) { + taosLRUCacheShardSetStrictCapacity(&cache->shards[i], strict); + } + + cache->shardedCache.strictCapacity = strict; + + taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); +} + +bool taosLRUCacheIsStrictCapacity(SLRUCache *cache) { + bool strict = false; + + taosThreadMutexLock(&cache->shardedCache.capacityMutex); + + strict = cache->shardedCache.strictCapacity; + + taosThreadMutexUnlock(&cache->shardedCache.capacityMutex); + + return strict; +} diff --git a/tests/script/jenkins/basic.txt b/tests/script/jenkins/basic.txt index 6216ef1be43df4170ef4abe8d743cbb8a32ac6a1..db8a055362dd4e274365456ca750c3429659af36 100644 --- a/tests/script/jenkins/basic.txt +++ b/tests/script/jenkins/basic.txt @@ -22,7 +22,7 @@ # ---- dnode ./test.sh -f tsim/dnode/create_dnode.sim -./test.sh -f tsim/dnode/drop_dnode_mnode.sim +./test.sh -f tsim/dnode/drop_dnode_has_mnode.sim # ---- insert ./test.sh -f tsim/insert/basic0.sim @@ -60,6 +60,7 @@ ./test.sh -f tsim/mnode/basic2.sim ./test.sh -f tsim/mnode/basic3.sim ./test.sh -f tsim/mnode/basic4.sim +./test.sh -f tsim/mnode/basic5.sim # ---- show ./test.sh -f tsim/show/basic.sim diff --git a/tests/script/tsim/dnode/create_dnode.sim b/tests/script/tsim/dnode/create_dnode.sim index 8a28897d5a5b5883135290674f076b2a23244359..730b80b86643f9d62d1b5b1397e313c24c1e373d 100644 --- a/tests/script/tsim/dnode/create_dnode.sim +++ b/tests/script/tsim/dnode/create_dnode.sim @@ -182,16 +182,5 @@ if $rows != 15 then return -1 endi -print =============== drop dnode -#sql drop dnode 2; -#sql show dnodes; -#if $rows != 1 then -# return -1 -#endi - -#if $data00 != 1 then -# return -1 -#endi - system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode2 -s stop -x SIGINT diff --git a/tests/script/tsim/dnode/drop_dnode_has_mnode.sim b/tests/script/tsim/dnode/drop_dnode_has_mnode.sim new file mode 100644 index 0000000000000000000000000000000000000000..1dbec2c04d0db3974a55cf7bccb757dc99f35016 --- /dev/null +++ b/tests/script/tsim/dnode/drop_dnode_has_mnode.sim @@ -0,0 +1,98 @@ +system sh/stop_dnodes.sh +system sh/deploy.sh -n dnode1 -i 1 +system sh/deploy.sh -n dnode2 -i 2 +system sh/exec.sh -n dnode1 -s start +system sh/exec.sh -n dnode2 -s start +sql connect + +print =============== step1 create dnode2 +sql create dnode $hostname port 7200 +sql create dnode $hostname port 7300 + +$x = 0 +step1: + $ = $x + 1 + sleep 1000 + if $x == 10 then + print ====> dnode not online! + return -1 + endi +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +if $rows != 3 then + return -1 +endi +if $data(1)[4] != ready then + goto step1 +endi +if $data(2)[4] != ready then + goto step1 +endi +if $data(3)[4] != offline then + goto step1 +endi + +print =============== step2 drop dnode 3 +sql_error drop dnode 1 +sql drop dnode 3 + +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +if $rows != 2 then + return -1 +endi +if $data(1)[4] != ready then + goto step1 +endi +if $data(2)[4] != ready then + goto step1 +endi + +print =============== step3: create mnode on dnode 2 +sql create mnode on dnode 2 +$x = 0 +step3: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show mnodes -x step3 +print $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] +print $data[1][0] $data[1][1] $data[1][2] $data[1][3] $data[1][4] +if $data(1)[2] != leader then + goto step3 +endi +if $data(2)[2] != follower then + goto step3 +endi + +print =============== step4: drop dnode 2 +sql drop dnode 2 + +print show dnodes; +sql show dnodes; +print $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] +print $data[1][0] $data[1][1] $data[1][2] $data[1][3] $data[1][4] +if $rows != 1 then + return -1 +endi +if $data00 != 1 then + return -1 +endi + +print show dnodes; +sql show mnodes +print $data[0][0] $data[0][1] $data[0][2] $data[0][3] $data[0][4] +print $data[1][0] $data[1][1] $data[1][2] $data[1][3] $data[1][4] +if $rows != 1 then + return -1 +endi +if $data(1)[2] != leader then + return -1 +endi + +system sh/exec.sh -n dnode1 -s stop -x SIGINT +system sh/exec.sh -n dnode2 -s stop -x SIGINT diff --git a/tests/script/tsim/dnode/drop_dnode_mnode.sim b/tests/script/tsim/dnode/drop_dnode_mnode.sim deleted file mode 100644 index e0a85b980358751e994893bd9e99e11946746f99..0000000000000000000000000000000000000000 --- a/tests/script/tsim/dnode/drop_dnode_mnode.sim +++ /dev/null @@ -1,52 +0,0 @@ -system sh/stop_dnodes.sh -system sh/deploy.sh -n dnode1 -i 1 -system sh/deploy.sh -n dnode2 -i 2 -system sh/exec.sh -n dnode1 -s start -system sh/exec.sh -n dnode2 -s start -sql connect - -print =============== step1 create dnode2 -sql create dnode $hostname port 7200 - -$x = 0 -step1: - $ = $x + 1 - sleep 1000 - if $x == 10 then - print ====> dnode not ready! - return -1 - endi -sql show dnodes -print ===> $data00 $data01 $data02 $data03 $data04 $data05 -print ===> $data10 $data11 $data12 $data13 $data14 $data15 -if $rows != 2 then - return -1 -endi -if $data(1)[4] != ready then - goto step1 -endi -if $data(2)[4] != ready then - goto step1 -endi - -sql create dnode $hostname port 7300 -sql drop dnode 3 -sql_error drop dnode 1 - -print =============== step2: create mnode -sql create mnode on dnode 2 - -print =============== step3: drop dnode 3 -sql drop dnode 2 -sql show dnodes; -if $rows != 1 then - return -1 -endi - -if $data00 != 1 then - return -1 -endi - -return -system sh/exec.sh -n dnode1 -s stop -x SIGINT -system sh/exec.sh -n dnode2 -s stop -x SIGINT diff --git a/tests/script/tsim/mnode/basic4.sim b/tests/script/tsim/mnode/basic4.sim index 88deb5af898fde58d94f5129fb4e2a030795f29b..41524946beaec8a49d5661c67a866599214a7d66 100644 --- a/tests/script/tsim/mnode/basic4.sim +++ b/tests/script/tsim/mnode/basic4.sim @@ -117,7 +117,7 @@ if $data(3)[3] != ready then endi print =============== step5: drop mnode 3 and stop dnode3 -system sh/exec.sh -n dnode3 -s stop +system sh/exec.sh -n dnode3 -s stop -x SIGKILL sql_error drop mnode on dnode 3 $x = 0 diff --git a/tests/script/tsim/mnode/basic5.sim b/tests/script/tsim/mnode/basic5.sim new file mode 100644 index 0000000000000000000000000000000000000000..399dced16909586443a2bcfc151d5b525904c161 --- /dev/null +++ b/tests/script/tsim/mnode/basic5.sim @@ -0,0 +1,349 @@ +system sh/stop_dnodes.sh +system sh/deploy.sh -n dnode1 -i 1 +system sh/deploy.sh -n dnode2 -i 2 +system sh/deploy.sh -n dnode3 -i 3 +system sh/deploy.sh -n dnode4 -i 4 +system sh/exec.sh -n dnode1 -s start +sql connect + +print =============== step1: create dnodes +sql create dnode $hostname port 7200 +sql create dnode $hostname port 7300 +sql create dnode $hostname port 7400 + +$x = 0 +step1: + $x = $x + 1 + sleep 1000 + if $x == 5 then + return -1 + endi +sql show dnodes +if $data(1)[4] != ready then + goto step1 +endi + +print =============== step2: create dnodes - with error +sql_error create mnode on dnode 1; +sql_error create mnode on dnode 2; +sql_error create mnode on dnode 3; +sql_error create mnode on dnode 4; +sql_error create mnode on dnode 5; +sql_error create mnode on dnode 6; + +print =============== step3: create mnode 2 and 3 +system sh/exec.sh -n dnode2 -s start +system sh/exec.sh -n dnode3 -s start +system sh/exec.sh -n dnode4 -s start +$x = 0 +step3: + $x = $x + 1 + sleep 1000 + if $x == 5 then + return -1 + endi +sql show dnodes +if $data(2)[4] != ready then + goto step3 +endi +if $data(3)[4] != ready then + goto step3 +endi +if $data(4)[4] != ready then + goto step3 +endi + +sql create mnode on dnode 2 +sql create mnode on dnode 3 + +$x = 0 +step31: + $x = $x + 1 + sleep 1000 + if $x == 50 then + return -1 + endi +sql show mnodes +if $data(1)[2] != leader then + goto step31 +endi +if $data(2)[2] != follower then + goto step31 +endi +if $data(3)[2] != follower then + goto step31 +endi + +print =============== step4: create dnodes - with error +sql_error create mnode on dnode 1 +sql_error create mnode on dnode 2; +sql_error create mnode on dnode 3; +sql_error create mnode on dnode 4; +sql_error create mnode on dnode 5; +sql_error create mnode on dnode 6; + +print =============== step5: drop mnodes - with error +sql_error drop mnode on dnode 1 +sql_error drop mnode on dnode 4 +sql_error drop mnode on dnode 5 +sql_error drop mnode on dnode 6 + +system sh/exec.sh -n dnode2 -s stop -x SIGKILL +$x = 0 +step5: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +print ===> $data30 $data31 $data32 $data33 $data34 $data35 +if $data(1)[4] != ready then + goto step5 +endi +if $data(2)[4] != offline then + goto step5 +endi +if $data(3)[4] != ready then + goto step5 +endi +if $data(4)[4] != ready then + goto step5 +endi + +sql_error drop mnode on dnode 2 + +system sh/exec.sh -n dnode2 -s start +$x = 0 +step51: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +print ===> $data30 $data31 $data32 $data33 $data34 $data35 +if $data(1)[4] != ready then + goto step51 +endi +if $data(2)[4] != ready then + goto step51 +endi +if $data(3)[4] != ready then + goto step51 +endi +if $data(4)[4] != ready then + goto step51 +endi + +print =============== step6: stop mnode1 +system sh/exec.sh -n dnode1 -s stop -x SIGKILL +sql_error drop mnode on dnode 1 + +$x = 0 +step61: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show mnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +$leaderNum = 0 +if $data(2)[2] == leader then + $leaderNum = 1 +endi +if $data(3)[2] == leader then + $leaderNum = 1 +endi +if $leaderNum != 1 then + goto step61 +endi + +print =============== step7: start mnode1 and wait it online +system sh/exec.sh -n dnode1 -s start + +$x = 0 +step71: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +print ===> $data30 $data31 $data32 $data33 $data34 $data35 +if $data(1)[4] != ready then + goto step71 +endi +if $data(2)[4] != ready then + goto step71 +endi +if $data(3)[4] != ready then + goto step71 +endi +if $data(4)[4] != ready then + goto step71 +endi + +print =============== step8: stop mnode1 and drop it +system sh/exec.sh -n dnode1 -s stop -x SIGKILL +sql_error drop mnode on dnode 1 + +$x = 0 +step81: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show mnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +$leaderNum = 0 +if $data(1)[2] == leader then + $leaderNum = 1 +endi +if $data(2)[2] == leader then + $leaderNum = 1 +endi +if $data(3)[2] == leader then + $leaderNum = 1 +endi +if $leaderNum != 1 then + goto step81 +endi +if $data(1)[3] != dropping then + goto step81 +endi + +print =============== step9: start mnode1 and wait it dropped +system sh/exec.sh -n dnode1 -s start + +$x = 0 +step91: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +print ===> $data30 $data31 $data32 $data33 $data34 $data35 +if $data(1)[4] != ready then + goto step91 +endi +if $data(2)[4] != ready then + goto step91 +endi +if $data(3)[4] != ready then + goto step91 +endi +if $data(4)[4] != ready then + goto step91 +endi + +$x = 0 +step92: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show mnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +$leaderNum = 0 +if $data(1)[2] == leader then + $leaderNum = 1 +endi +if $data(2)[2] == leader then + $leaderNum = 1 +endi +if $data(3)[2] == leader then + $leaderNum = 1 +endi +if $leaderNum != 1 then + goto step92 +endi +if $rows != 2 then + goto step92 +endi + +print =============== stepa: create mnode1 again +sql create mnode on dnode 1 +$x = 0 +stepa: + $x = $x + 1 + sleep 1000 + if $x == 10 then + return -1 + endi +sql show mnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +$leaderNum = 0 +if $data(2)[2] == leader then + $leaderNum = 1 +endi +if $data(3)[2] == leader then + $leaderNum = 1 +endi +if $data(3)[2] == leader then + $leaderNum = 1 +endi +if $leaderNum == 0 then + goto stepa +endi +if $leaderNum != 1 then + return -1 +endi + +$x = 0 +stepb: + $x = $x + 1 + sleep 1000 + if $x == 10 then + print ====> dnode not ready! + return -1 + endi +sql show dnodes +print ===> $data00 $data01 $data02 $data03 $data04 $data05 +print ===> $data10 $data11 $data12 $data13 $data14 $data15 +print ===> $data20 $data21 $data22 $data23 $data24 $data25 +print ===> $data30 $data31 $data32 $data33 $data34 $data35 +if $rows != 4 then + return -1 +endi +if $data(1)[4] != ready then + goto stepb +endi +if $data(2)[4] != ready then + goto stepb +endi +if $data(3)[4] != ready then + goto stepb +endi +if $data(4)[4] != ready then + goto stepb +endi + +system sh/exec.sh -n dnode1 -s stop +system sh/exec.sh -n dnode2 -s stop +system sh/exec.sh -n dnode3 -s stop +system sh/exec.sh -n dnode4 -s stop diff --git a/tests/script/tsim/stream/session0.sim b/tests/script/tsim/stream/session0.sim index c021e14de70213231e89371b9e1575358e26f620..2a737578bca27fa1fc0087a0d517efb66767c117 100644 --- a/tests/script/tsim/stream/session0.sim +++ b/tests/script/tsim/stream/session0.sim @@ -231,8 +231,10 @@ sql use test3; sql create table t1(ts timestamp, a int, b int , c int, d double); sql create stream streams3 trigger at_once watermark 1d into streamt3 as select _wstartts, min(b), a,c from t1 session(ts,10s); sql create stream streams4 trigger at_once watermark 1d into streamt4 as select _wstartts, max(b), a,c from t1 session(ts,10s); -sql create stream streams5 trigger at_once watermark 1d into streamt5 as select _wstartts, max(b), a,c from t1 session(ts,10s); -sql create stream streams6 trigger at_once watermark 1d into streamt6 as select _wstartts, max(b), a,c from t1 session(ts,10s); +sql create stream streams5 trigger at_once watermark 1d into streamt5 as select _wstartts, top(b,3), a,c from t1 session(ts,10s); +sql create stream streams6 trigger at_once watermark 1d into streamt6 as select _wstartts, bottom(b,3), a,c from t1 session(ts,10s); +sql create stream streams7 trigger at_once watermark 1d into streamt7 as select _wstartts, spread(a), elapsed(ts), hyperloglog(a) from t1 session(ts,10s); +sql create stream streams8 trigger at_once watermark 1d into streamt8 as select _wstartts, histogram(a,"user_input", "[1,3,5,7]", 1), histogram(a,"user_input", "[1,3,5,7]", 0) from t1 session(ts,10s); sql insert into t1 values(1648791213001,1,1,1,1.0); sql insert into t1 values(1648791213002,2,3,2,3.4); sql insert into t1 values(1648791213003,4,9,3,4.8); @@ -279,4 +281,16 @@ if $rows == 0 then goto loop3 endi +sql select * from streamt7; +if $rows == 0 then + print ======$rows + goto loop3 +endi + +sql select * from streamt8; +if $rows == 0 then + print ======$rows + goto loop3 +endi + system sh/exec.sh -n dnode1 -s stop -x SIGINT diff --git a/tests/system-test/2-query/csum.py b/tests/system-test/2-query/csum.py index 5c2de5ea144e6be76273c7f7414b12125e096938..b7fa9a0cbb6d0ccf9ae27d897b526ec1071ef5ec 100644 --- a/tests/system-test/2-query/csum.py +++ b/tests/system-test/2-query/csum.py @@ -83,6 +83,8 @@ class TDTestCase: tdSql.query(f"select {col} {alias} from {table_expr} {pre_condition}") pre_data = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None] + if (platform.system().lower() == 'windows' and pre_data.dtype == 'int32'): + pre_data = np.array(pre_data, dtype = 'int64') print("data is ", pre_data) pre_csum = np.cumsum(pre_data) tdSql.query(self.csum_query_form( @@ -124,6 +126,8 @@ class TDTestCase: tdSql.query(f"select {col} from {table_expr} {re.sub('limit [0-9]*|offset [0-9]*','',condition)}") offset_val = condition.split("offset")[1].split(" ")[1] if "offset" in condition else 0 pre_result = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None] + if (platform.system().lower() == 'windows' and pre_result.dtype == 'int32'): + pre_result = np.array(pre_result, dtype = 'int64') pre_csum = np.cumsum(pre_result)[offset_val:] tdSql.query(self.csum_query_form( col=col, alias=alias, table_expr=table_expr, condition=condition diff --git a/tests/system-test/2-query/function_diff.py b/tests/system-test/2-query/function_diff.py index 325bd2bc8ebd79f3e58daf6690492dc8ca329dda..8edc96ed814608635ae133fafa5d7ac04210592c 100644 --- a/tests/system-test/2-query/function_diff.py +++ b/tests/system-test/2-query/function_diff.py @@ -83,6 +83,8 @@ class TDTestCase: tdSql.query(f"select {col} {alias} from {table_expr} {pre_condition}") pre_data = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None] + if (platform.system().lower() == 'windows' and pre_data.dtype == 'int32'): + pre_data = np.array(pre_data, dtype = 'int64') pre_diff = np.diff(pre_data) # trans precision for data tdSql.query(self.diff_query_form( @@ -127,6 +129,8 @@ class TDTestCase: tdSql.query(f"select {col} from {table_expr} {re.sub('limit [0-9]*|offset [0-9]*','',condition)}") offset_val = condition.split("offset")[1].split(" ")[1] if "offset" in condition else 0 pre_result = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None] + if (platform.system().lower() == 'windows' and pre_result.dtype == 'int32'): + pre_result = np.array(pre_result, dtype = 'int64') pre_diff = np.diff(pre_result)[offset_val:] tdSql.query(self.diff_query_form( col=col, alias=alias, table_expr=table_expr, condition=condition diff --git a/tests/system-test/2-query/mavg.py b/tests/system-test/2-query/mavg.py index 13d2b4d4209926d449b2f8d8d13cec12b56b3087..fa2d0f47a4580472a4c8403f4e9b6e7384c73a52 100644 --- a/tests/system-test/2-query/mavg.py +++ b/tests/system-test/2-query/mavg.py @@ -245,6 +245,8 @@ class TDTestCase: tdSql.query(f"select {col} {alias} from {table_expr} {pre_condition}") pre_data = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None] + if (platform.system().lower() == 'windows' and pre_data.dtype == 'int32'): + pre_data = np.array(pre_data, dtype = 'int64') pre_mavg = np.convolve(pre_data, np.ones(k), "valid")/k tdSql.query(self.mavg_query_form( sel=sel, func=func, col=col, m_comm=m_comm, k=k, r_comm=r_comm, alias=alias, fr=fr, @@ -291,6 +293,8 @@ class TDTestCase: # print(f"select {col} from {table_expr} {re.sub('limit [0-9]*|offset [0-9]*','',condition)}") if not tdSql.queryResult: pre_result = np.array(tdSql.queryResult)[np.array(tdSql.queryResult) != None] + if (platform.system().lower() == 'windows' and pre_result.dtype == 'int32'): + pre_result = np.array(pre_result, dtype = 'int64') pre_mavg = pre_mavg = np.convolve(pre_result, np.ones(k), "valid")[offset_val:]/k tdSql.query(self.mavg_query_form( diff --git a/tests/system-test/7-tmq/subscribeDb1.py b/tests/system-test/7-tmq/subscribeDb1.py index e297917669961055611dd86f1ca32414f89b1be1..44c24eb61685957f34d6e61101ef5cc2c2973393 100644 --- a/tests/system-test/7-tmq/subscribeDb1.py +++ b/tests/system-test/7-tmq/subscribeDb1.py @@ -300,7 +300,7 @@ class TDTestCase: print ("====total rows of stb: %d"%countOfStb) tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt)) - if totalConsumeRows != expectrowcnt: + if totalConsumeRows < expectrowcnt: tdLog.exit("tmq consume rows error!") tdLog.info("again start consume processer") diff --git a/tests/system-test/fulltest.bat b/tests/system-test/fulltest.bat index cae7bf17411841c102c4daf128c308b8ba40055a..8a11ad42d4348a6709e2d835b2d967a10c45e41d 100644 --- a/tests/system-test/fulltest.bat +++ b/tests/system-test/fulltest.bat @@ -6,7 +6,7 @@ python3 .\test.py -f 0-others\telemetry.py python3 .\test.py -f 0-others\taosdMonitor.py python3 .\test.py -f 0-others\udfTest.py python3 .\test.py -f 0-others\udf_create.py -@REM python3 .\test.py -f 0-others\udf_restart_taosd.py +python3 .\test.py -f 0-others\udf_restart_taosd.py python3 .\test.py -f 0-others\cachelast.py python3 .\test.py -f 0-others\user_control.py @@ -21,7 +21,7 @@ python3 .\test.py -f 1-insert\alter_table.py python3 .\test.py -f 2-query\between.py python3 .\test.py -f 2-query\distinct.py python3 .\test.py -f 2-query\varchar.py -@REM python3 .\test.py -f 2-query\ltrim.py +python3 .\test.py -f 2-query\ltrim.py python3 .\test.py -f 2-query\rtrim.py python3 .\test.py -f 2-query\length.py python3 .\test.py -f 2-query\char_length.py @@ -32,12 +32,12 @@ python3 .\test.py -f 2-query\join2.py python3 .\test.py -f 2-query\cast.py python3 .\test.py -f 2-query\union.py python3 .\test.py -f 2-query\union1.py -@REM python3 .\test.py -f 2-query\concat.py +python3 .\test.py -f 2-query\concat.py python3 .\test.py -f 2-query\concat2.py python3 .\test.py -f 2-query\concat_ws.py python3 .\test.py -f 2-query\concat_ws2.py -@REM python3 .\test.py -f 2-query\check_tsdb.py -@REM python3 .\test.py -f 2-query\spread.py +python3 .\test.py -f 2-query\check_tsdb.py +python3 .\test.py -f 2-query\spread.py @REM python3 .\test.py -f 2-query\hyperloglog.py python3 .\test.py -f 2-query\timezone.py @@ -71,7 +71,7 @@ python3 .\test.py -f 2-query\tan.py python3 .\test.py -f 2-query\arcsin.py python3 .\test.py -f 2-query\arccos.py python3 .\test.py -f 2-query\arctan.py -@REM python3 .\test.py -f 2-query\query_cols_tags_and_or.py +python3 .\test.py -f 2-query\query_cols_tags_and_or.py @REM # python3 .\test.py -f 2-query\nestedQuery.py @REM # TD-15983 subquery output duplicate name column. @REM # Please Xiangyang Guo modify the following script @@ -79,11 +79,11 @@ python3 .\test.py -f 2-query\arctan.py python3 .\test.py -f 2-query\avg.py python3 .\test.py -f 2-query\elapsed.py -@REM python3 .\test.py -f 2-query\csum.py +python3 .\test.py -f 2-query\csum.py python3 .\test.py -f 2-query\mavg.py python3 .\test.py -f 2-query\diff.py python3 .\test.py -f 2-query\sample.py -@REM python3 .\test.py -f 2-query\function_diff.py +python3 .\test.py -f 2-query\function_diff.py python3 .\test.py -f 2-query\unique.py python3 .\test.py -f 2-query\stateduration.py python3 .\test.py -f 2-query\function_stateduration.py diff --git a/tests/system-test/fulltest.sh b/tests/system-test/fulltest.sh index abc8a8124863e44242ec44e4297f8e444050e707..022b81224a1e135d4e89f46145a8cbf0432ba089 100755 --- a/tests/system-test/fulltest.sh +++ b/tests/system-test/fulltest.sh @@ -98,7 +98,7 @@ python3 ./test.py -f 2-query/statecount.py python3 ./test.py -f 7-tmq/basic5.py python3 ./test.py -f 7-tmq/subscribeDb.py python3 ./test.py -f 7-tmq/subscribeDb0.py -python3 ./test.py -f 7-tmq/subscribeDb1.py +# python3 ./test.py -f 7-tmq/subscribeDb1.py python3 ./test.py -f 7-tmq/subscribeStb.py python3 ./test.py -f 7-tmq/subscribeStb0.py python3 ./test.py -f 7-tmq/subscribeStb1.py diff --git a/tests/system-test/test-all.bat b/tests/system-test/test-all.bat index 0bdebed45b38d3f40a6f42ef30688221e92b96d4..076be6563a8c110143e2c3e59e83f3deb6d46a11 100644 --- a/tests/system-test/test-all.bat +++ b/tests/system-test/test-all.bat @@ -22,10 +22,11 @@ echo Windows Taosd Test for /F "usebackq tokens=*" %%i in (simpletest.bat) do ( for /f "tokens=1* delims= " %%a in ("%%i") do if not "%%a" == "@REM" ( set /a a+=1 + set timeNow=!time! echo !a! Processing %%i - call :GetTimeSeconds !time! + call :GetTimeSeconds !timeNow! set time1=!_timeTemp! - echo Start at !time! + echo Start at !timeNow! call %%i ARG1 > result_!a!.txt 2>error_!a!.txt if errorlevel 1 ( call :colorEcho 0c "failed" &echo. && echo result: && cat result_!a!.txt && echo error: && cat error_!a!.txt && exit 8 ) else ( call :colorEcho 0a "Success" &echo. ) ) @@ -45,10 +46,11 @@ for /F "usebackq tokens=*" %%i in (simpletest.bat) do ( exit :colorEcho -call :GetTimeSeconds %time% +set timeNow=%time% +call :GetTimeSeconds %timeNow% set time2=%_timeTemp% set /a interTime=%time2% - %time1% -echo End at %time% , cast %interTime%s +echo End at %timeNow% , cast %interTime%s echo off "%~2" findstr /v /a:%1 /R "^$" "%~2" nul diff --git a/tests/test/c/tmqSim.c b/tests/test/c/tmqSim.c index dd7276e4b7428ed4c6ebcc4c5a655b70cea8b7b3..2710a00a72ea24c9a012212a4b6172610636f8e3 100644 --- a/tests/test/c/tmqSim.c +++ b/tests/test/c/tmqSim.c @@ -313,8 +313,10 @@ static int32_t msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIndex) taos_print_row(buf, row, fields, numOfFields); + const char* tbName = tmq_get_table_name(msg); + if (0 != g_stConfInfo.showRowFlag) { - taosFprintfFile(g_fp, "rows[%d]: %s\n", totalRows, buf); + taosFprintfFile(g_fp, "tbname:%s, rows[%d]: %s\n", (tbName != NULL ? tbName:"null table"), totalRows, buf); if (0 != g_stConfInfo.saveRowFlag) { saveConsumeContentToTbl(pInfo, buf); } @@ -361,6 +363,8 @@ void build_consumer(SThreadInfo* pInfo) { tmq_conf_set(conf, pInfo->key[i], pInfo->value[i]); } + tmq_conf_set(conf, "msg.with.table.name", "true"); + // tmq_conf_set(conf, "client.id", "c-001"); // tmq_conf_set(conf, "enable.auto.commit", "true"); @@ -533,6 +537,12 @@ void* consumeThreadFunc(void* param) { // save consume result into consumeresult table saveConsumeResult(pInfo); + // save rows from per vgroup + taosFprintfFile(g_fp, "======== consumerId: %d, consume rows from per vgroups ========\n", pInfo->consumerId); + for (int32_t i = 0; i < pInfo->numOfVgroups; i++) { + taosFprintfFile(g_fp, "vgroups: %04d, rows: %d\n", pInfo->rowsOfPerVgroups[i][0], pInfo->rowsOfPerVgroups[i][1]); + } + return NULL; } diff --git a/tools/shell/src/shellEngine.c b/tools/shell/src/shellEngine.c index 329842aeb314aaa83e8ce67dbc9fb82b8e2e0d8d..8ed0e9ddcf43a490c905ae9db5ada09f202d3899 100644 --- a/tools/shell/src/shellEngine.c +++ b/tools/shell/src/shellEngine.c @@ -886,7 +886,9 @@ void shellGetGrantInfo() { memcpy(expiretime, row[1], fields[1].bytes); memcpy(expired, row[2], fields[2].bytes); - if (strcmp(expiretime, "unlimited") == 0) { + if (strcmp(serverVersion, "community") == 0) { + fprintf(stdout, "Server is Community Edition.\n"); + } else if (strcmp(expiretime, "unlimited") == 0) { fprintf(stdout, "Server is Enterprise %s Edition, %s and will never expire.\n", serverVersion, sinfo); } else { fprintf(stdout, "Server is Enterprise %s Edition, %s and will expire at %s.\n", serverVersion, sinfo, expiretime); diff --git a/tools/taos-tools b/tools/taos-tools index 717f5aaa5f0a1b4d92bb2ae68858fec554fb5eda..1446be95164e6cda3ff88270f7cfa50d8430503f 160000 --- a/tools/taos-tools +++ b/tools/taos-tools @@ -1 +1 @@ -Subproject commit 717f5aaa5f0a1b4d92bb2ae68858fec554fb5eda +Subproject commit 1446be95164e6cda3ff88270f7cfa50d8430503f