diff --git a/include/util/tdef.h b/include/util/tdef.h index 71fb91dbcb12b331b9446d93d8331989c5d4c522..a4d88806a9b6eb76714a1d1546544e892921621f 100644 --- a/include/util/tdef.h +++ b/include/util/tdef.h @@ -253,6 +253,7 @@ typedef enum ELogicConditionType { #define TSDB_IPv4ADDR_LEN 16 #define TSDB_FILENAME_LEN 128 #define TSDB_SHOW_SQL_LEN 2048 +#define TSDB_SHOW_SCHEMA_JSON_LEN TSDB_MAX_COLUMNS * 256 #define TSDB_SLOW_QUERY_SQL_LEN 512 #define TSDB_SHOW_SUBQUERY_LEN 1000 diff --git a/source/common/src/systable.c b/source/common/src/systable.c index 062bae68c83206d7e99e9ec3f8198ab7ec35961f..db0cc78de682a948126fd476f4129406c50740b8 100644 --- a/source/common/src/systable.c +++ b/source/common/src/systable.c @@ -280,7 +280,9 @@ static const SSysDbTableSchema topicSchema[] = { {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = false}, {.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, - // TODO config + {.name = "schema", .bytes = TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, + {.name = "meta", .bytes = 4 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, + {.name = "type", .bytes = 8 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, }; diff --git a/source/dnode/mnode/impl/inc/mndDb.h b/source/dnode/mnode/impl/inc/mndDb.h index 97d047d7a3db51b0893ad2fd4a45c76cf1ed8424..4411cb4785a5c4613f03e4d2173c1f9bb524dc85 100644 --- a/source/dnode/mnode/impl/inc/mndDb.h +++ b/source/dnode/mnode/impl/inc/mndDb.h @@ -32,6 +32,7 @@ bool mndIsDbReady(SMnode *pMnode, SDbObj *pDb); SSdbRaw *mndDbActionEncode(SDbObj *pDb); const char *mndGetDbStr(const char *src); +const char *mndGetStableStr(const char *src); int32_t mndProcessCompactDbReq(SRpcMsg *pReq); diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 8d1a811f142976cfb375a78518dfc6d8430c9d68..aee8872393263bec2367f346b763c440dae3ad23 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -521,6 +521,7 @@ typedef struct { char* physicalPlan; SSchemaWrapper schema; int64_t stbUid; + char stbName[TSDB_TABLE_FNAME_LEN]; // forbid condition int64_t ntbUid; SArray* ntbColIds; diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c index df220340688ae7b0230dae3a3242e0fc9ff6d2b9..de23eddbf9b83ba2fd9145f3379f753125e64c47 100644 --- a/source/dnode/mnode/impl/src/mndDb.c +++ b/source/dnode/mnode/impl/src/mndDb.c @@ -1513,6 +1513,13 @@ const char *mndGetDbStr(const char *src) { return pos; } +const char *mndGetStableStr(const char *src) { + char *pos = strstr(src, TS_PATH_DELIMITER); + if (pos != NULL) ++pos; + if (pos == NULL) return src; + return mndGetDbStr(pos); +} + static int64_t getValOfDiffPrecision(int8_t unit, int64_t val) { int64_t v = 0; switch (unit) { diff --git a/source/dnode/mnode/impl/src/mndDump.c b/source/dnode/mnode/impl/src/mndDump.c index a991bddda8a997c6074ee21e811f3829e0370fe8..d57053bb5bd1d26827728264277323183b996a3c 100644 --- a/source/dnode/mnode/impl/src/mndDump.c +++ b/source/dnode/mnode/impl/src/mndDump.c @@ -285,6 +285,7 @@ void dumpTopic(SSdb *pSdb, SJson *json) { tjsonAddStringToObject(item, "subType", i642str(pObj->subType)); tjsonAddStringToObject(item, "withMeta", i642str(pObj->withMeta)); tjsonAddStringToObject(item, "stbUid", i642str(pObj->stbUid)); + tjsonAddStringToObject(item, "stbName", mndGetStableStr(pObj->stbName)); tjsonAddStringToObject(item, "sqlLen", i642str(pObj->sqlLen)); tjsonAddStringToObject(item, "astLen", i642str(pObj->astLen)); tjsonAddStringToObject(item, "sqlLen", i642str(pObj->sqlLen)); diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index f6da3709164158acc008708121873da5bb2b78d1..b3f7176a9356a7e76ea91084292a223afcc06ffa 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -109,6 +109,7 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { SDB_SET_INT8(pRaw, dataPos, pTopic->withMeta, TOPIC_ENCODE_OVER); SDB_SET_INT64(pRaw, dataPos, pTopic->stbUid, TOPIC_ENCODE_OVER); + SDB_SET_BINARY(pRaw, dataPos, pTopic->stbName, TSDB_TABLE_FNAME_LEN, TOPIC_ENCODE_OVER); SDB_SET_INT32(pRaw, dataPos, pTopic->sqlLen, TOPIC_ENCODE_OVER); SDB_SET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_ENCODE_OVER); SDB_SET_INT32(pRaw, dataPos, pTopic->astLen, TOPIC_ENCODE_OVER); @@ -196,6 +197,7 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) { SDB_GET_INT8(pRaw, dataPos, &pTopic->withMeta, TOPIC_DECODE_OVER); SDB_GET_INT64(pRaw, dataPos, &pTopic->stbUid, TOPIC_DECODE_OVER); + SDB_GET_BINARY(pRaw, dataPos, pTopic->stbName, TSDB_TABLE_FNAME_LEN, TOPIC_DECODE_OVER); SDB_GET_INT32(pRaw, dataPos, &pTopic->sqlLen, TOPIC_DECODE_OVER); pTopic->sql = taosMemoryCalloc(pTopic->sqlLen, sizeof(char)); if (pTopic->sql == NULL) { @@ -460,6 +462,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq * return -1; } + strcpy(topicObj.stbName, pCreate->subStbName); topicObj.stbUid = pStb->uid; mndReleaseStb(pMnode, pStb); } @@ -830,6 +833,43 @@ int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) { return 0; } +static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){ + char* string = NULL; + cJSON* columns = cJSON_CreateArray(); + if (columns == NULL) { + return; + } + for (int i = 0; i < nCols; i++) { + cJSON* column = cJSON_CreateObject(); + SSchema* s = schema + i; + cJSON* cname = cJSON_CreateString(s->name); + cJSON_AddItemToObject(column, "name", cname); + cJSON* ctype = cJSON_CreateString(tDataTypes[s->type].name); + cJSON_AddItemToObject(column, "type", ctype); + int32_t length = 0; + if (s->type == TSDB_DATA_TYPE_BINARY) { + length = s->bytes - VARSTR_HEADER_SIZE; + } else if (s->type == TSDB_DATA_TYPE_NCHAR || s->type == TSDB_DATA_TYPE_JSON) { + length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE; + } else{ + length = s->bytes; + } + cJSON* cbytes = cJSON_CreateNumber(length); + cJSON_AddItemToObject(column, "length", cbytes); + cJSON_AddItemToArray(columns, column); + } + string = cJSON_PrintUnformatted(columns); + cJSON_Delete(columns); + + size_t len = strlen(string); + if(string && len <= TSDB_SHOW_SCHEMA_JSON_LEN){ + STR_TO_VARSTR(schemaJson, string); + }else{ + mError("mndRetrieveTopic build schema error json:%p, json len:%zu", string, len); + } + taosMemoryFree(string); +} + static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) { SMnode *pMnode = pReq->info.node; SSdb *pSdb = pMnode->pSdb; @@ -868,6 +908,47 @@ static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataSetVal(pColInfo, numOfRows, (const char *)sql, false); + char schemaJson[TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE] = {0}; + if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){ + schemaToJson(pTopic->schema.pSchema, pTopic->schema.nCols, schemaJson); + }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){ + SStbObj *pStb = mndAcquireStb(pMnode, pTopic->stbName); + if (pStb == NULL) { + terrno = TSDB_CODE_MND_STB_NOT_EXIST; + return -1; + } + schemaToJson(pStb->pColumns, pStb->numOfColumns, schemaJson); + + mndReleaseStb(pMnode, pStb); + }else{ + STR_TO_VARSTR(schemaJson, "NULL"); + } + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)schemaJson, false); + + char mete[4 + VARSTR_HEADER_SIZE] = {0}; + if(pTopic->withMeta){ + STR_TO_VARSTR(mete, "yes"); + }else{ + STR_TO_VARSTR(mete, "no"); + } + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)mete, false); + + char type[8 + VARSTR_HEADER_SIZE] = {0}; + if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){ + STR_TO_VARSTR(type, "column"); + }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){ + STR_TO_VARSTR(type, "stable"); + }else{ + STR_TO_VARSTR(type, "db"); + } + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataSetVal(pColInfo, numOfRows, (const char *)type, false); + numOfRows++; sdbRelease(pSdb, pTopic); } diff --git a/tests/system-test/7-tmq/ins_topics_test.py b/tests/system-test/7-tmq/ins_topics_test.py new file mode 100644 index 0000000000000000000000000000000000000000..54b6c141f55ddbe7fbb29f0f1ba5c81b39ffd1c3 --- /dev/null +++ b/tests/system-test/7-tmq/ins_topics_test.py @@ -0,0 +1,59 @@ + +import taos +import sys +import time +import socket +import os +import threading + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * + +class TDTestCase: + hostname = socket.gethostname() + # rpcDebugFlagVal = '143' + #clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''} + #clientCfgDict["rpcDebugFlag"] = rpcDebugFlagVal + #updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''} + # updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal + #print ("===================: ", updatecfgDict) + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor()) + #tdSql.init(conn.cursor(), logSql) # output sql.txt file + + def test(self): + tdLog.info("create database, stb, ctb") + tdSql.execute("create database if not exists db1 vgroups 4 wal_retention_period 3600") + tdSql.execute("create table if not exists db1.st(ts timestamp, c1 int, c2 bool, c3 tinyint, c4 double, c5 nchar(8)) tags(t1 int, t2 float, t3 binary(4))") + tdSql.execute("create table if not exists db1.nt(ts timestamp, c1 smallint, c2 float, c3 binary(64), c4 bigint)") + tdSql.execute("create table if not exists db1.st1 using db1.st tags(1, 9.3, \"st1\")") + + tdLog.info("create topic") + tdSql.execute("create topic topic_1 as database db1") + tdSql.execute("create topic topic_2 as stable db1.st") + tdSql.execute("create topic topic_3 as select * from db1.nt") + tdSql.execute("create topic topic_4 as select ts,c3,c5 from db1.st") + + tdSql.query("select * from information_schema.ins_topics") + tdSql.checkRows(4) + # tdSql.checkData(0, 1, 51) + # tdSql.checkData(0, 4, 940) + # tdSql.checkData(1, 1, 23) + # tdSql.checkData(1, 4, None) + + tdLog.printNoPrefix("======== test case end ...... ") + + def run(self): + self.test() + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase())