提交 495e417c 编写于 作者: S Shengliang Guan

enh<cluster>: add show message funcs

上级 9a004f68
...@@ -119,6 +119,11 @@ extern const int32_t TYPE_BYTES[15]; ...@@ -119,6 +119,11 @@ extern const int32_t TYPE_BYTES[15];
#define TSDB_INS_TABLE_BNODES "bnodes" #define TSDB_INS_TABLE_BNODES "bnodes"
#define TSDB_INS_TABLE_SNODES "snodes" #define TSDB_INS_TABLE_SNODES "snodes"
#define TSDB_INS_TABLE_LICENCES "grants" #define TSDB_INS_TABLE_LICENCES "grants"
#define TSDB_INS_TABLE_TRANS "trans"
#define TSDB_INS_TABLE_SMAS "smas"
#define TSDB_INS_TABLE_CONFIGS "configs"
#define TSDB_INS_TABLE_CONNS "connections"
#define TSDB_INS_TABLE_QUERIES "queries"
#define TSDB_INDEX_TYPE_SMA "SMA" #define TSDB_INDEX_TYPE_SMA "SMA"
#define TSDB_INDEX_TYPE_FULLTEXT "FULLTEXT" #define TSDB_INDEX_TYPE_FULLTEXT "FULLTEXT"
...@@ -498,6 +503,15 @@ enum { ...@@ -498,6 +503,15 @@ enum {
#define QNODE_HANDLE 1 #define QNODE_HANDLE 1
#define DEFAULT_HANDLE 0 #define DEFAULT_HANDLE 0
#define TSDB_CONFIG_OPTION_LEN 16
#define TSDB_CONIIG_VALUE_LEN 48
#define TSDB_CONFIG_NUMBER 8
#define QUERY_ID_SIZE 20
#define QUERY_OBJ_ID_SIZE 18
#define SUBQUERY_INFO_SIZE 6
#define QUERY_SAVE_SIZE 20
#define MAX_NUM_STR_SIZE 40 #define MAX_NUM_STR_SIZE 40
......
...@@ -2184,7 +2184,6 @@ int32_t tSerializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableReq ...@@ -2184,7 +2184,6 @@ int32_t tSerializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableReq
if (tStartEncode(&encoder) < 0) return -1; if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeI64(&encoder, pReq->showId) < 0) return -1; if (tEncodeI64(&encoder, pReq->showId) < 0) return -1;
if (tEncodeI32(&encoder, pReq->type) < 0) return -1; if (tEncodeI32(&encoder, pReq->type) < 0) return -1;
// if (tEncodeI8(&encoder, pReq->free) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->db) < 0) return -1; if (tEncodeCStr(&encoder, pReq->db) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->tb) < 0) return -1; if (tEncodeCStr(&encoder, pReq->tb) < 0) return -1;
tEndEncode(&encoder); tEndEncode(&encoder);
...@@ -2201,7 +2200,6 @@ int32_t tDeserializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableR ...@@ -2201,7 +2200,6 @@ int32_t tDeserializeSRetrieveTableReq(void *buf, int32_t bufLen, SRetrieveTableR
if (tStartDecode(&decoder) < 0) return -1; if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeI64(&decoder, &pReq->showId) < 0) return -1; if (tDecodeI64(&decoder, &pReq->showId) < 0) return -1;
if (tDecodeI32(&decoder, &pReq->type) < 0) return -1; if (tDecodeI32(&decoder, &pReq->type) < 0) return -1;
// if (tDecodeI8(&decoder, &pReq->free) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1; if (tDecodeCStrTo(&decoder, pReq->db) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->tb) < 0) return -1; if (tDecodeCStrTo(&decoder, pReq->tb) < 0) return -1;
tEndDecode(&decoder); tEndDecode(&decoder);
......
...@@ -26,8 +26,6 @@ int32_t mndInitShow(SMnode *pMnode); ...@@ -26,8 +26,6 @@ int32_t mndInitShow(SMnode *pMnode);
void mndCleanupShow(SMnode *pMnode); void mndCleanupShow(SMnode *pMnode);
void mndAddShowRetrieveHandle(SMnode *pMnode, EShowType showType, ShowRetrieveFp fp); void mndAddShowRetrieveHandle(SMnode *pMnode, EShowType showType, ShowRetrieveFp fp);
void mndAddShowFreeIterHandle(SMnode *pMnode, EShowType msgType, ShowFreeIterFp fp); void mndAddShowFreeIterHandle(SMnode *pMnode, EShowType msgType, ShowFreeIterFp fp);
void mndVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capacity, SShowObj *pShow);
char *mndShowStr(int32_t showType);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
...@@ -24,9 +24,6 @@ ...@@ -24,9 +24,6 @@
#define TSDB_DNODE_VER_NUMBER 1 #define TSDB_DNODE_VER_NUMBER 1
#define TSDB_DNODE_RESERVE_SIZE 64 #define TSDB_DNODE_RESERVE_SIZE 64
#define TSDB_CONFIG_OPTION_LEN 16
#define TSDB_CONIIG_VALUE_LEN 48
#define TSDB_CONFIG_NUMBER 8
static const char *offlineReason[] = { static const char *offlineReason[] = {
"", "",
...@@ -55,7 +52,6 @@ static int32_t mndProcessConfigDnodeReq(SNodeMsg *pReq); ...@@ -55,7 +52,6 @@ static int32_t mndProcessConfigDnodeReq(SNodeMsg *pReq);
static int32_t mndProcessConfigDnodeRsp(SNodeMsg *pRsp); static int32_t mndProcessConfigDnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessStatusReq(SNodeMsg *pReq); static int32_t mndProcessStatusReq(SNodeMsg *pReq);
static int32_t mndGetConfigMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConfigs(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows); static int32_t mndRetrieveConfigs(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows);
static void mndCancelGetNextConfig(SMnode *pMnode, void *pIter); static void mndCancelGetNextConfig(SMnode *pMnode, void *pIter);
static int32_t mndRetrieveDnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows); static int32_t mndRetrieveDnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows);
...@@ -638,37 +634,6 @@ static int32_t mndProcessConfigDnodeRsp(SNodeMsg *pRsp) { ...@@ -638,37 +634,6 @@ static int32_t mndProcessConfigDnodeRsp(SNodeMsg *pRsp) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mndGetConfigMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_CONIIG_VALUE_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "value");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = TSDB_CONFIG_NUMBER;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveConfigs(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) { static int32_t mndRetrieveConfigs(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
int32_t totalRows = 0; int32_t totalRows = 0;
......
...@@ -20,7 +20,6 @@ ...@@ -20,7 +20,6 @@
#define SYSTABLE_SCH_DB_NAME_LEN ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE) #define SYSTABLE_SCH_DB_NAME_LEN ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
#define SYSTABLE_SCH_COL_NAME_LEN ((TSDB_COL_NAME_LEN - 1) + VARSTR_HEADER_SIZE) #define SYSTABLE_SCH_COL_NAME_LEN ((TSDB_COL_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
//!!!! Note: only APPEND columns in below tables, NO insert !!!!
static const SInfosTableSchema dnodesSchema[] = { static const SInfosTableSchema dnodesSchema[] = {
{.name = "id", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT}, {.name = "id", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT},
{.name = "endpoint", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "endpoint", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
...@@ -225,6 +224,54 @@ static const SInfosTableSchema subscribeSchema[] = { ...@@ -225,6 +224,54 @@ static const SInfosTableSchema subscribeSchema[] = {
{.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
}; };
static const SInfosTableSchema smaSchema[] = {
{.name = "sma_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "stable_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
};
static const SInfosTableSchema transSchema[] = {
{.name = "id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "created_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "stage", .bytes = TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "db", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "type", .bytes = TSDB_TRANS_TYPE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "last_exec_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "last_error", .bytes = (TSDB_TRANS_ERROR_LEN - 1) + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
};
static const SInfosTableSchema configSchema[] = {
{.name = "name", .bytes = TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "value", .bytes = TSDB_CONIIG_VALUE_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
};
static const SInfosTableSchema connSchema[] = {
{.name = "connId", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "user", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "program", .bytes = TSDB_APP_NAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "ip:port", .bytes = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "login_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "last_access", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
};
static const SInfosTableSchema querySchema[] = {
{.name = "queryId", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "connId", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "user", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "ip:port", .bytes = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "qid", .bytes = 22 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "created_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
{.name = "time", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
{.name = "sql_obj_id", .bytes = QUERY_OBJ_ID_SIZE + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "ep", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "stable_query", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL},
{.name = "sub_queries", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
{.name = "sub_query_info", .bytes = TSDB_SHOW_SUBQUERY_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
{.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
};
static const SInfosTableMeta infosMeta[] = { static const SInfosTableMeta infosMeta[] = {
{TSDB_INS_TABLE_DNODES, dnodesSchema, tListLen(dnodesSchema)}, {TSDB_INS_TABLE_DNODES, dnodesSchema, tListLen(dnodesSchema)},
{TSDB_INS_TABLE_MNODES, mnodesSchema, tListLen(mnodesSchema)}, {TSDB_INS_TABLE_MNODES, mnodesSchema, tListLen(mnodesSchema)},
...@@ -243,6 +290,11 @@ static const SInfosTableMeta infosMeta[] = { ...@@ -243,6 +290,11 @@ static const SInfosTableMeta infosMeta[] = {
{TSDB_INS_TABLE_USER_USERS, userUsersSchema, tListLen(userUsersSchema)}, {TSDB_INS_TABLE_USER_USERS, userUsersSchema, tListLen(userUsersSchema)},
{TSDB_INS_TABLE_LICENCES, grantsSchema, tListLen(grantsSchema)}, {TSDB_INS_TABLE_LICENCES, grantsSchema, tListLen(grantsSchema)},
{TSDB_INS_TABLE_VGROUPS, vgroupsSchema, tListLen(vgroupsSchema)}, {TSDB_INS_TABLE_VGROUPS, vgroupsSchema, tListLen(vgroupsSchema)},
{TSDB_INS_TABLE_TRANS, transSchema, tListLen(transSchema)},
{TSDB_INS_TABLE_SMAS, smaSchema, tListLen(smaSchema)},
{TSDB_INS_TABLE_CONFIGS, configSchema, tListLen(configSchema)},
{TSDB_INS_TABLE_CONNS, connSchema, tListLen(connSchema)},
{TSDB_INS_TABLE_QUERIES, querySchema, tListLen(querySchema)},
}; };
static int32_t mndInitInfosTableSchema(const SInfosTableSchema *pSrc, int32_t colNum, SSchema **pDst) { static int32_t mndInitInfosTableSchema(const SInfosTableSchema *pSrc, int32_t colNum, SSchema **pDst) {
......
...@@ -23,11 +23,6 @@ ...@@ -23,11 +23,6 @@
#include "tglobal.h" #include "tglobal.h"
#include "version.h" #include "version.h"
#define QUERY_ID_SIZE 20
#define QUERY_OBJ_ID_SIZE 18
#define SUBQUERY_INFO_SIZE 6
#define QUERY_SAVE_SIZE 20
typedef struct { typedef struct {
int32_t id; int32_t id;
int8_t connType; int8_t connType;
...@@ -56,9 +51,7 @@ static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq); ...@@ -56,9 +51,7 @@ static int32_t mndProcessHeartBeatReq(SNodeMsg *pReq);
static int32_t mndProcessConnectReq(SNodeMsg *pReq); static int32_t mndProcessConnectReq(SNodeMsg *pReq);
static int32_t mndProcessKillQueryReq(SNodeMsg *pReq); static int32_t mndProcessKillQueryReq(SNodeMsg *pReq);
static int32_t mndProcessKillConnReq(SNodeMsg *pReq); static int32_t mndProcessKillConnReq(SNodeMsg *pReq);
static int32_t mndGetConnsMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows); static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static int32_t mndGetQueryMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows); static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter); static void mndCancelGetNextQuery(SMnode *pMnode, void *pIter);
...@@ -558,81 +551,6 @@ static int32_t mndProcessKillConnReq(SNodeMsg *pReq) { ...@@ -558,81 +551,6 @@ static int32_t mndProcessKillConnReq(SNodeMsg *pReq) {
} }
} }
static int32_t mndGetConnsMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pReq->user);
if (pUser == NULL) return 0;
if (!pUser->superUser) {
mndReleaseUser(pMnode, pUser);
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
mndReleaseUser(pMnode, pUser);
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "connId");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "user");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
// app name
pShow->bytes[cols] = TSDB_APP_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "program");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
// app pid
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "pid");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ip:port");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "login_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "last_access");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = taosCacheGetNumOfObj(pMgmt->cache);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) { static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
int32_t numOfRows = 0; int32_t numOfRows = 0;
...@@ -692,121 +610,6 @@ static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int ...@@ -692,121 +610,6 @@ static int32_t mndRetrieveConns(SNodeMsg *pReq, SShowObj *pShow, char *data, int
return numOfRows; return numOfRows;
} }
static int32_t mndGetQueryMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SProfileMgmt *pMgmt = &pMnode->profileMgmt;
SUserObj *pUser = mndAcquireUser(pMnode, pReq->user);
if (pUser == NULL) return 0;
if (!pUser->superUser) {
mndReleaseUser(pMnode, pUser);
terrno = TSDB_CODE_MND_NO_RIGHTS;
return -1;
}
mndReleaseUser(pMnode, pUser);
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "queryId");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "connId");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_USER_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "user");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_IPv4ADDR_LEN + 6 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ip:port");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 22 + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "qid");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "created_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_BIGINT;
strcpy(pSchema[cols].name, "time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = QUERY_OBJ_ID_SIZE + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql_obj_id");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "pid");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_EP_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "ep");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 1;
pSchema[cols].type = TSDB_DATA_TYPE_BOOL;
strcpy(pSchema[cols].name, "stable_query");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "sub_queries");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_SHOW_SUBQUERY_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sub_query_info");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = 1000000;
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) { static int32_t mndRetrieveQueries(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
int32_t numOfRows = 0; int32_t numOfRows = 0;
......
...@@ -115,6 +115,7 @@ static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove) { ...@@ -115,6 +115,7 @@ static void mndReleaseShowObj(SShowObj *pShow, bool forceRemove) {
static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
SShowMgmt *pMgmt = &pMnode->showMgmt; SShowMgmt *pMgmt = &pMnode->showMgmt;
SShowObj *pShow = NULL;
int32_t rowsToRead = SHOW_STEP_SIZE; int32_t rowsToRead = SHOW_STEP_SIZE;
int32_t size = 0; int32_t size = 0;
int32_t rowsRead = 0; int32_t rowsRead = 0;
...@@ -125,8 +126,6 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -125,8 +126,6 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
return -1; return -1;
} }
SShowObj* pShow = NULL;
if (retrieveReq.showId == 0) { if (retrieveReq.showId == 0) {
SShowReq req = {0}; SShowReq req = {0};
req.type = retrieveReq.type; req.type = retrieveReq.type;
...@@ -143,7 +142,7 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -143,7 +142,7 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
pShow->numOfColumns = pShow->pMeta->numOfColumns; pShow->numOfColumns = pShow->pMeta->numOfColumns;
int32_t offset = 0; int32_t offset = 0;
for(int32_t i = 0; i < pShow->pMeta->numOfColumns; ++i) { for (int32_t i = 0; i < pShow->pMeta->numOfColumns; ++i) {
pShow->offset[i] = offset; pShow->offset[i] = offset;
int32_t bytes = pShow->pMeta->pSchemas[i].bytes; int32_t bytes = pShow->pMeta->pSchemas[i].bytes;
...@@ -162,26 +161,25 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -162,26 +161,25 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type]; ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type];
if (retrieveFp == NULL) { if (retrieveFp == NULL) {
mndReleaseShowObj((SShowObj*) pShow, false); mndReleaseShowObj(pShow, false);
terrno = TSDB_CODE_MSG_NOT_PROCESSED; terrno = TSDB_CODE_MSG_NOT_PROCESSED;
mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr()); mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr());
return -1; return -1;
} }
mDebug("show:0x%" PRIx64 ", start retrieve data, type:%s", pShow->id, mndShowStr(pShow->type)); mDebug("show:0x%" PRIx64 ", start retrieve data, type:%d", pShow->id, pShow->type);
int32_t numOfCols = pShow->pMeta->numOfColumns;
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock)); int32_t numOfCols = pShow->pMeta->numOfColumns;
SSDataBlock *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)); pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
pBlock->info.numOfCols = numOfCols; pBlock->info.numOfCols = numOfCols;
for(int32_t i = 0; i < numOfCols; ++i) { for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {0}; SColumnInfoData idata = {0};
SSchema* p = &pShow->pMeta->pSchemas[i]; SSchema *p = &pShow->pMeta->pSchemas[i];
idata.info.bytes = p->bytes; idata.info.bytes = p->bytes;
idata.info.type = p->type; idata.info.type = p->type;
idata.info.colId = p->colId; idata.info.colId = p->colId;
taosArrayPush(pBlock->pDataBlock, &idata); taosArrayPush(pBlock->pDataBlock, &idata);
...@@ -191,15 +189,15 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -191,15 +189,15 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
} }
blockDataEnsureCapacity(pBlock, rowsToRead); blockDataEnsureCapacity(pBlock, rowsToRead);
if (mndCheckRetrieveFinished((SShowObj*) pShow)) { if (mndCheckRetrieveFinished(pShow)) {
mDebug("show:0x%" PRIx64 ", read finished, numOfRows:%d", pShow->id, pShow->numOfRows); mDebug("show:0x%" PRIx64 ", read finished, numOfRows:%d", pShow->id, pShow->numOfRows);
rowsRead = 0; rowsRead = 0;
} else { } else {
rowsRead = (*retrieveFp)(pReq, (SShowObj *)pShow, pBlock, rowsToRead); rowsRead = (*retrieveFp)(pReq, pShow, pBlock, rowsToRead);
if (rowsRead < 0) { if (rowsRead < 0) {
terrno = rowsRead; terrno = rowsRead;
mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id); mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id);
mndReleaseShowObj((SShowObj *)pShow, true); mndReleaseShowObj(pShow, true);
return -1; return -1;
} }
...@@ -207,13 +205,12 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -207,13 +205,12 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
mDebug("show:0x%" PRIx64 ", stop retrieve data, rowsRead:%d numOfRows:%d", pShow->id, rowsRead, pShow->numOfRows); mDebug("show:0x%" PRIx64 ", stop retrieve data, rowsRead:%d numOfRows:%d", pShow->id, rowsRead, pShow->numOfRows);
} }
// numOfCols + sizeof(SSysTableSchema) * numOfCols + data payload size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pShow->pMeta->numOfColumns +
size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pShow->pMeta->numOfColumns + blockDataGetSize(pBlock) blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(pBlock);
+ blockDataGetSerialMetaSize(pBlock);
SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size); SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
if (pRsp == NULL) { if (pRsp == NULL) {
mndReleaseShowObj((SShowObj*) pShow, false); mndReleaseShowObj(pShow, false);
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr()); mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr());
blockDataDestroy(pBlock); blockDataDestroy(pBlock);
...@@ -222,9 +219,8 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -222,9 +219,8 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
pRsp->handle = htobe64(pShow->id); pRsp->handle = htobe64(pShow->id);
// if free flag is set, client wants to clean the resources
if (rowsRead > 0) { if (rowsRead > 0) {
char * pStart = pRsp->data; char *pStart = pRsp->data;
SSchema *ps = pShow->pMeta->pSchemas; SSchema *ps = pShow->pMeta->pSchemas;
*(int32_t *)pStart = htonl(pShow->pMeta->numOfColumns); *(int32_t *)pStart = htonl(pShow->pMeta->numOfColumns);
...@@ -245,77 +241,22 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) { ...@@ -245,77 +241,22 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
pRsp->numOfRows = htonl(rowsRead); pRsp->numOfRows = htonl(rowsRead);
pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision
pReq->pRsp = pRsp; pReq->pRsp = pRsp;
pReq->rspLen = size; pReq->rspLen = size;
if (rowsRead == 0 || rowsRead < rowsToRead) { if (rowsRead == 0 || rowsRead < rowsToRead) {
pRsp->completed = 1; pRsp->completed = 1;
mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id); mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id);
mndReleaseShowObj((SShowObj*) pShow, true); mndReleaseShowObj(pShow, true);
} else { } else {
mDebug("show:0x%" PRIx64 ", retrieve not completed yet", pShow->id); mDebug("show:0x%" PRIx64 ", retrieve not completed yet", pShow->id);
mndReleaseShowObj((SShowObj*) pShow, false); mndReleaseShowObj(pShow, false);
} }
blockDataDestroy(pBlock); blockDataDestroy(pBlock);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
char *mndShowStr(int32_t showType) {
switch (showType) {
case TSDB_MGMT_TABLE_ACCT:
return "show accounts";
case TSDB_MGMT_TABLE_USER:
return "show users";
case TSDB_MGMT_TABLE_DB:
return "show databases";
case TSDB_MGMT_TABLE_TABLE:
return "show tables";
case TSDB_MGMT_TABLE_DNODE:
return "show dnodes";
case TSDB_MGMT_TABLE_MNODE:
return "show mnodes";
case TSDB_MGMT_TABLE_QNODE:
return "show qnodes";
case TSDB_MGMT_TABLE_SNODE:
return "show snodes";
case TSDB_MGMT_TABLE_BNODE:
return "show bnodes";
case TSDB_MGMT_TABLE_VGROUP:
return "show vgroups";
case TSDB_MGMT_TABLE_STB:
return "show stables";
case TSDB_MGMT_TABLE_MODULE:
return "show modules";
case TSDB_MGMT_TABLE_QUERIES:
return "show queries";
case TSDB_MGMT_TABLE_STREAMS:
return "show streams";
case TSDB_MGMT_TABLE_VARIABLES:
return "show configs";
case TSDB_MGMT_TABLE_CONNS:
return "show connections";
case TSDB_MGMT_TABLE_TRANS:
return "show trans";
case TSDB_MGMT_TABLE_GRANTS:
return "show grants";
case TSDB_MGMT_TABLE_VNODES:
return "show vnodes";
case TSDB_MGMT_TABLE_CLUSTER:
return "show cluster";
case TSDB_MGMT_TABLE_STREAMTABLES:
return "show streamtables";
case TSDB_MGMT_TABLE_TP:
return "show topics";
case TSDB_MGMT_TABLE_FUNC:
return "show functions";
case TSDB_MGMT_TABLE_INDEX:
return "show indexes";
default:
return "undefined";
}
}
static bool mndCheckRetrieveFinished(SShowObj *pShow) { static bool mndCheckRetrieveFinished(SShowObj *pShow) {
if (pShow->pIter == NULL && pShow->numOfRows != 0) { if (pShow->pIter == NULL && pShow->numOfRows != 0) {
return true; return true;
...@@ -323,14 +264,6 @@ static bool mndCheckRetrieveFinished(SShowObj *pShow) { ...@@ -323,14 +264,6 @@ static bool mndCheckRetrieveFinished(SShowObj *pShow) {
return false; return false;
} }
void mndVacuumResult(char *data, int32_t numOfCols, int32_t rows, int32_t capacity, SShowObj *pShow) {
if (rows < capacity) {
for (int32_t i = 0; i < numOfCols; ++i) {
memmove(data + pShow->offset[i] * rows, data + pShow->offset[i] * capacity, pShow->bytes[i] * rows);
}
}
}
void mndAddShowRetrieveHandle(SMnode *pMnode, EShowType showType, ShowRetrieveFp fp) { void mndAddShowRetrieveHandle(SMnode *pMnode, EShowType showType, ShowRetrieveFp fp) {
SShowMgmt *pMgmt = &pMnode->showMgmt; SShowMgmt *pMgmt = &pMnode->showMgmt;
pMgmt->retrieveFps[showType] = fp; pMgmt->retrieveFps[showType] = fp;
......
...@@ -724,46 +724,6 @@ static int32_t mndProcessVDropSmaRsp(SNodeMsg *pRsp) { ...@@ -724,46 +724,6 @@ static int32_t mndProcessVDropSmaRsp(SNodeMsg *pRsp) {
return 0; return 0;
} }
static int32_t mndGetSmaMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_INDEX_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "stb");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = sdbGetSize(pSdb, SDB_SMA);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveSma(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) { static int32_t mndRetrieveSma(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
......
...@@ -58,7 +58,7 @@ int32_t mndInitStream(SMnode *pMnode) { ...@@ -58,7 +58,7 @@ int32_t mndInitStream(SMnode *pMnode) {
/*mndSetMsgHandle(pMnode, TDMT_MND_DROP_STREAM, mndProcessDropStreamReq);*/ /*mndSetMsgHandle(pMnode, TDMT_MND_DROP_STREAM, mndProcessDropStreamReq);*/
/*mndSetMsgHandle(pMnode, TDMT_MND_DROP_STREAM_RSP, mndProcessDropStreamInRsp);*/ /*mndSetMsgHandle(pMnode, TDMT_MND_DROP_STREAM_RSP, mndProcessDropStreamInRsp);*/
// mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TP, mndRetrieveStream); // mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TP, mndRetrieveStream);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TP, mndCancelGetNextStream); mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TP, mndCancelGetNextStream);
return sdbSetTable(pMnode->pSdb, table); return sdbSetTable(pMnode->pSdb, table);
...@@ -421,50 +421,6 @@ static int32_t mndGetNumOfStreams(SMnode *pMnode, char *dbName, int32_t *pNumOfS ...@@ -421,50 +421,6 @@ static int32_t mndGetNumOfStreams(SMnode *pMnode, char *dbName, int32_t *pNumOfS
return 0; return 0;
} }
static int32_t mndGetStreamMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
if (mndGetNumOfStreams(pMnode, pShow->db, &pShow->numOfRows) != 0) {
return -1;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = sdbGetSize(pSdb, SDB_STREAM);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveStream(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) { static int32_t mndRetrieveStream(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
......
...@@ -35,7 +35,6 @@ static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pTopic, SMqTopicObj ...@@ -35,7 +35,6 @@ static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pTopic, SMqTopicObj
static int32_t mndProcessCreateTopicReq(SNodeMsg *pReq); static int32_t mndProcessCreateTopicReq(SNodeMsg *pReq);
static int32_t mndProcessDropTopicReq(SNodeMsg *pReq); static int32_t mndProcessDropTopicReq(SNodeMsg *pReq);
static int32_t mndProcessDropTopicInRsp(SNodeMsg *pRsp); static int32_t mndProcessDropTopicInRsp(SNodeMsg *pRsp);
static int32_t mndGetTopicMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows); static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter); static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter);
...@@ -500,50 +499,6 @@ static int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTo ...@@ -500,50 +499,6 @@ static int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTo
return 0; return 0;
} }
static int32_t mndGetTopicMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
if (mndGetNumOfTopics(pMnode, pShow->db, &pShow->numOfRows) != 0) {
return -1;
}
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "name");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "sql");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = sdbGetSize(pSdb, SDB_TOPIC);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) { static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
......
...@@ -58,7 +58,6 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans); ...@@ -58,7 +58,6 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
static int32_t mndProcessTransReq(SNodeMsg *pReq); static int32_t mndProcessTransReq(SNodeMsg *pReq);
static int32_t mndProcessKillTransReq(SNodeMsg *pReq); static int32_t mndProcessKillTransReq(SNodeMsg *pReq);
static int32_t mndGetTransMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta);
static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows); static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows);
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter); static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
...@@ -1260,69 +1259,6 @@ void mndTransPullup(SMnode *pMnode) { ...@@ -1260,69 +1259,6 @@ void mndTransPullup(SMnode *pMnode) {
sdbWriteFile(pMnode->pSdb); sdbWriteFile(pMnode->pSdb);
} }
static int32_t mndGetTransMeta(SNodeMsg *pReq, SShowObj *pShow, STableMetaRsp *pMeta) {
SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb;
int32_t cols = 0;
SSchema *pSchema = pMeta->pSchemas;
pShow->bytes[cols] = 4;
pSchema[cols].type = TSDB_DATA_TYPE_INT;
strcpy(pSchema[cols].name, "id");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "create_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "stage");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = (TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "db");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = TSDB_TRANS_TYPE_LEN + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "type");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = 8;
pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
strcpy(pSchema[cols].name, "last_exec_time");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pShow->bytes[cols] = (TSDB_TRANS_ERROR_LEN - 1) + VARSTR_HEADER_SIZE;
pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
strcpy(pSchema[cols].name, "last_error");
pSchema[cols].bytes = pShow->bytes[cols];
cols++;
pMeta->numOfColumns = cols;
pShow->numOfColumns = cols;
pShow->offset[0] = 0;
for (int32_t i = 1; i < cols; ++i) {
pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
}
pShow->numOfRows = sdbGetSize(pSdb, SDB_TRANS);
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
strcpy(pMeta->tbName, mndShowStr(pShow->type));
return 0;
}
static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) { static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pReq->pNode; SMnode *pMnode = pReq->pNode;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
......
...@@ -5,13 +5,13 @@ add_subdirectory(acct) ...@@ -5,13 +5,13 @@ add_subdirectory(acct)
#add_subdirectory(trans) #add_subdirectory(trans)
#add_subdirectory(qnode) #add_subdirectory(qnode)
#add_subdirectory(snode) #add_subdirectory(snode)
#add_subdirectory(bnode) add_subdirectory(bnode)
#add_subdirectory(show) #add_subdirectory(show)
#add_subdirectory(profile) #add_subdirectory(profile)
add_subdirectory(dnode) #add_subdirectory(dnode)
#add_subdirectory(mnode) #add_subdirectory(mnode)
add_subdirectory(db) #add_subdirectory(db)
add_subdirectory(stb) #add_subdirectory(stb)
#add_subdirectory(sma) #add_subdirectory(sma)
#add_subdirectory(func) #add_subdirectory(func)
#add_subdirectory(topic) #add_subdirectory(topic)
aux_source_directory(. STEST_SRC) aux_source_directory(. MNODE_BNODE_TEST_SRC)
add_executable(mnode_test_bnode ${STEST_SRC}) add_executable(mbnodeTest ${MNODE_BNODE_TEST_SRC})
target_link_libraries( target_link_libraries(
mnode_test_bnode mbnodeTest
PUBLIC sut PUBLIC sut
) )
add_test( add_test(
NAME mnode_test_bnode NAME mbnodeTest
COMMAND mnode_test_bnode COMMAND mbnodeTest
) )
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册