xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

add sub structure

This commit is contained in:
Liu Jicong 2022-01-04 10:40:34 +08:00
parent 7f80c1e969
commit 0a83ecac81
4 changed files with 264 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

104
include/common/tmsg.h
View File

@ -320,14 +320,23 @@ typedef struct SEpSet {
} SEpSet; } SEpSet;
static FORCE_INLINE int taosEncodeSEpSet(void** buf, const SEpSet* pEp) { static FORCE_INLINE int taosEncodeSEpSet(void** buf, const SEpSet* pEp) {
if(buf == NULL) return sizeof(SEpSet); int tlen = 0;
memcpy(buf, pEp, sizeof(SEpSet)); tlen += taosEncodeFixedI8(buf, pEp->inUse);
//TODO: endian conversion tlen += taosEncodeFixedI8(buf, pEp->numOfEps);
return sizeof(SEpSet); for (int i = 0; i < TSDB_MAX_REPLICA; i++) {
tlen += taosEncodeFixedU16(buf, pEp->port[i]);
tlen += taosEncodeString(buf, pEp->fqdn[i]);
}
return tlen;
} }
static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEpSet) { static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEp) {
memcpy(pEpSet, buf, sizeof(SEpSet)); buf = taosDecodeFixedI8(buf, &pEp->inUse);
buf = taosDecodeFixedI8(buf, &pEp->numOfEps);
for (int i = 0; i < TSDB_MAX_REPLICA; i++) {
buf = taosDecodeFixedU16(buf, &pEp->port[i]);
buf = taosDecodeStringTo(buf, pEp->fqdn[i]);
}
return buf; return buf;
} }
@ -1088,16 +1097,16 @@ typedef struct STaskDropRsp {
} STaskDropRsp; } STaskDropRsp;
typedef struct { typedef struct {
int8_t igExists; int8_t igExists;
char* name; char* name;
char* physicalPlan; char* physicalPlan;
char* logicalPlan; char* logicalPlan;
} SCMCreateTopicReq; } SCMCreateTopicReq;
static FORCE_INLINE int tSerializeSCMCreateTopicReq(void** buf, const SCMCreateTopicReq* pReq) { static FORCE_INLINE int tSerializeSCMCreateTopicReq(void** buf, const SCMCreateTopicReq* pReq) {
int tlen = 0; int tlen = 0;
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeFixedI8(buf, pReq->igExists); tlen += taosEncodeFixedI8(buf, pReq->igExists);
tlen += taosEncodeString(buf, pReq->name);
tlen += taosEncodeString(buf, pReq->physicalPlan); tlen += taosEncodeString(buf, pReq->physicalPlan);
tlen += taosEncodeString(buf, pReq->logicalPlan); tlen += taosEncodeString(buf, pReq->logicalPlan);
return tlen; return tlen;
@ -1127,41 +1136,62 @@ static FORCE_INLINE void* tDeserializeSCMCreateTopicRsp(void* buf, SCMCreateTopi
} }
typedef struct { typedef struct {
char* topicName; int32_t topicNum;
char* consumerGroup;
int64_t consumerId; int64_t consumerId;
char* consumerGroup;
char* topicName[];
} SCMSubscribeReq; } SCMSubscribeReq;
static FORCE_INLINE int tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) { static FORCE_INLINE int tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
int tlen = 0; int tlen = 0;
tlen += taosEncodeString(buf, pReq->topicName); tlen += taosEncodeFixedI32(buf, pReq->topicNum);
tlen += taosEncodeString(buf, pReq->consumerGroup);
tlen += taosEncodeFixedI64(buf, pReq->consumerId); tlen += taosEncodeFixedI64(buf, pReq->consumerId);
tlen += taosEncodeString(buf, pReq->consumerGroup);
for(int i = 0; i < pReq->topicNum; i++) {
tlen += taosEncodeString(buf, pReq->topicName[i]);
}
return tlen; return tlen;
} }
static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) { static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
buf = taosDecodeString(buf, &pReq->topicName); buf = taosDecodeFixedI32(buf, &pReq->topicNum);
buf = taosDecodeString(buf, &pReq->consumerGroup);
buf = taosDecodeFixedI64(buf, &pReq->consumerId); buf = taosDecodeFixedI64(buf, &pReq->consumerId);
buf = taosDecodeString(buf, &pReq->consumerGroup);
for(int i = 0; i < pReq->topicNum; i++) {
buf = taosDecodeString(buf, &pReq->topicName[i]);
}
return buf; return buf;
} }
typedef struct { typedef struct SMqSubTopic {
int32_t vgId; int32_t vgId;
SEpSet pEpSet; int64_t topicId;
SEpSet epSet;
} SMqSubTopic;
typedef struct {
int32_t topicNum;
SMqSubTopic topics[];
} SCMSubscribeRsp; } SCMSubscribeRsp;
static FORCE_INLINE int tSerializeSCMSubscribeRsp(void** buf, const SCMSubscribeRsp* pRsp) { static FORCE_INLINE int tSerializeSCMSubscribeRsp(void** buf, const SCMSubscribeRsp* pRsp) {
int tlen = 0; int tlen = 0;
tlen += taosEncodeFixedI32(buf, pRsp->vgId); tlen += taosEncodeFixedI32(buf, pRsp->topicNum);
tlen += taosEncodeSEpSet(buf, &pRsp->pEpSet); for(int i = 0; i < pRsp->topicNum; i++) {
tlen += taosEncodeFixedI32(buf, pRsp->topics[i].vgId);
tlen += taosEncodeFixedI64(buf, pRsp->topics[i].topicId);
tlen += taosEncodeSEpSet(buf, &pRsp->topics[i].epSet);
}
return tlen; return tlen;
} }
static FORCE_INLINE void* tDeserializeSCMSubscribeRsp(void* buf, SCMSubscribeRsp* pRsp) { static FORCE_INLINE void* tDeserializeSCMSubscribeRsp(void* buf, SCMSubscribeRsp* pRsp) {
buf = taosDecodeFixedI32(buf, &pRsp->vgId); buf = taosDecodeFixedI32(buf, &pRsp->topicNum);
buf = taosDecodeSEpSet(buf, &pRsp->pEpSet); for(int i = 0; i < pRsp->topicNum; i++) {
buf = taosDecodeFixedI32(buf, &pRsp->topics[i].vgId);
buf = taosDecodeFixedI64(buf, &pRsp->topics[i].topicId);
buf = taosDecodeSEpSet(buf, &pRsp->topics[i].epSet);
}
return buf; return buf;
} }
@ -1170,10 +1200,36 @@ typedef struct {
int64_t consumerId; int64_t consumerId;
int64_t consumerGroupId; int64_t consumerGroupId;
int64_t offset; int64_t offset;
char* sql;
char* logicalPlan;
char* physicalPlan;
} SMVSubscribeReq; } SMVSubscribeReq;
static FORCE_INLINE int tSerializeSMVSubscribeReq(void* buf, SMVSubscribeReq* pReq) {
int tlen = 0;
tlen += taosEncodeFixedI64(buf, pReq->topicId);
tlen += taosEncodeFixedI64(buf, pReq->consumerId);
tlen += taosEncodeFixedI64(buf, pReq->consumerGroupId);
tlen += taosEncodeFixedI64(buf, pReq->offset);
tlen += taosEncodeString(buf, pReq->sql);
tlen += taosEncodeString(buf, pReq->logicalPlan);
tlen += taosEncodeString(buf, pReq->physicalPlan);
return tlen;
}
static FORCE_INLINE void* tDeserializeSMVSubscribeReq(void* buf, SMVSubscribeReq* pReq) {
buf = taosDecodeFixedI64(buf, &pReq->topicId);
buf = taosDecodeFixedI64(buf, &pReq->consumerId);
buf = taosDecodeFixedI64(buf, &pReq->consumerGroupId);
buf = taosDecodeFixedI64(buf, &pReq->offset);
buf = taosDecodeString(buf, &pReq->sql);
buf = taosDecodeString(buf, &pReq->logicalPlan);
buf = taosDecodeString(buf, &pReq->physicalPlan);
return buf;
}
typedef struct { typedef struct {
int64_t newOffset; int64_t status;
} SMVSubscribeRsp; } SMVSubscribeRsp;
typedef struct { typedef struct {

1
include/util/tdef.h
View File

@ -177,6 +177,7 @@ do { \
#define TSDB_TYPE_STR_MAX_LEN 32 #define TSDB_TYPE_STR_MAX_LEN 32
#define TSDB_TABLE_FNAME_LEN (TSDB_DB_FNAME_LEN + TSDB_TABLE_NAME_LEN + TSDB_NAME_DELIMITER_LEN) #define TSDB_TABLE_FNAME_LEN (TSDB_DB_FNAME_LEN + TSDB_TABLE_NAME_LEN + TSDB_NAME_DELIMITER_LEN)
#define TSDB_TOPIC_FNAME_LEN TSDB_TABLE_FNAME_LEN #define TSDB_TOPIC_FNAME_LEN TSDB_TABLE_FNAME_LEN
#define TSDB_CONSUMER_GROUP_LEN 192
#define TSDB_COL_NAME_LEN 65 #define TSDB_COL_NAME_LEN 65
#define TSDB_MAX_SAVED_SQL_LEN TSDB_MAX_COLUMNS * 64 #define TSDB_MAX_SAVED_SQL_LEN TSDB_MAX_COLUMNS * 64
#define TSDB_MAX_SQL_LEN TSDB_PAYLOAD_SIZE #define TSDB_MAX_SQL_LEN TSDB_PAYLOAD_SIZE

103
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -22,6 +22,7 @@
#include "sync.h" #include "sync.h"
#include "tmsg.h" #include "tmsg.h"
#include "thash.h" #include "thash.h"
#include "tlist.h"
#include "tlog.h" #include "tlog.h"
#include "trpc.h" #include "trpc.h"
#include "ttimer.h" #include "ttimer.h"
@ -288,14 +289,81 @@ typedef struct {
char payload[]; char payload[];
} SShowObj; } SShowObj;
typedef struct SConsumerObj {
uint64_t uid;
int64_t createTime;
int64_t updateTime;
//uint64_t dbUid;
int32_t version;
SRWLatch lock;
SList* topics;
} SConsumerObj;
typedef struct SMqSubConsumerObj {
int64_t consumerUid; // if -1, unassigned
SList* vgId; //SList<int32_t>
} SMqSubConsumerObj;
typedef struct SMqSubCGroupObj {
char name[TSDB_CONSUMER_GROUP_LEN];
SList* consumers; //SList<SMqConsumerObj>
} SMqSubCGroupObj;
typedef struct SMqSubTopicObj {
char name[TSDB_TOPIC_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN];
int64_t createTime;
int64_t updateTime;
int64_t uid;
int64_t dbUid;
int32_t version;
SRWLatch lock;
int32_t sqlLen;
char* sql;
char* logicalPlan;
char* physicalPlan;
SList* cgroups; //SList<SMqSubCGroupObj>
} SMqSubTopicObj;
typedef struct SMqConsumerSubObj {
int64_t topicUid;
SList* vgIds; //SList<int64_t>
} SMqConsumerSubObj;
typedef struct SMqConsumerHbObj {
int64_t consumerId;
SList* consumerSubs; //SList<SMqConsumerSubObj>
} SMqConsumerHbObj;
typedef struct SMqVGroupSubObj {
int64_t topicUid;
SList* consumerIds; //SList<int64_t>
} SMqVGroupSubObj;
typedef struct SMqVGroupHbObj {
int64_t vgId;
SList* vgSubs; //SList<SMqVGroupSubObj>
} SMqVGroupHbObj;
typedef struct SCGroupObj {
char name[TSDB_TOPIC_FNAME_LEN];
int64_t createTime;
int64_t updateTime;
uint64_t uid;
//uint64_t dbUid;
int32_t version;
SRWLatch lock;
SList* consumerIds;
} SCGroupObj;
typedef struct { typedef struct {
char name[TSDB_TOPIC_FNAME_LEN]; char name[TSDB_TOPIC_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN]; char db[TSDB_DB_FNAME_LEN];
int64_t createTime; int64_t createTime;
int64_t updateTime; int64_t updateTime;
uint64_t uid; uint64_t uid;
uint64_t dbUid; uint64_t dbUid;
int32_t version; int32_t version;
SRWLatch lock; SRWLatch lock;
int32_t execLen; int32_t execLen;
void* executor; void* executor;
@ -303,32 +371,9 @@ typedef struct {
char* sql; char* sql;
char* logicalPlan; char* logicalPlan;
char* physicalPlan; char* physicalPlan;
SList* consumerIds;
} STopicObj; } STopicObj;
typedef struct {
char name[TSDB_TOPIC_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN];
int64_t createTime;
int64_t updateTime;
uint64_t uid;
//uint64_t dbUid;
int32_t version;
SRWLatch lock;
} SConsumerObj;
typedef struct {
char name[TSDB_TOPIC_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN];
int64_t createTime;
int64_t updateTime;
uint64_t uid;
//uint64_t dbUid;
int32_t version;
SRWLatch lock;
} SCGroupObj;
typedef struct SMnodeMsg { typedef struct SMnodeMsg {
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];
char db[TSDB_DB_FNAME_LEN]; char db[TSDB_DB_FNAME_LEN];

140
source/dnode/mnode/impl/src/mndConsumer.c
View File

@ -66,17 +66,82 @@ int32_t mndInitConsumer(SMnode *pMnode) {
void mndCleanupConsumer(SMnode *pMnode) {} void mndCleanupConsumer(SMnode *pMnode) {}
static SSdbRaw *mndCGroupActionEncode(SCGroupObj *pCGroup) {
int32_t size = sizeof(SConsumerObj) + MND_CONSUMER_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);
if (pRaw == NULL) return NULL;
int32_t dataPos = 0;
SDB_SET_BINARY(pRaw, dataPos, pCGroup->name, TSDB_TABLE_FNAME_LEN);
SDB_SET_INT64(pRaw, dataPos, pCGroup->createTime);
SDB_SET_INT64(pRaw, dataPos, pCGroup->updateTime);
SDB_SET_INT64(pRaw, dataPos, pCGroup->uid);
/*SDB_SET_INT64(pRaw, dataPos, pConsumer->dbUid);*/
SDB_SET_INT32(pRaw, dataPos, pCGroup->version);
int32_t sz = listNEles(pCGroup->consumerIds);
SDB_SET_INT32(pRaw, dataPos, sz);
SListIter iter;
tdListInitIter(pCGroup->consumerIds, &iter, TD_LIST_FORWARD);
SListNode *pn = NULL;
while ((pn = tdListNext(&iter)) != NULL) {
int64_t consumerId = *(int64_t *)pn->data;
SDB_SET_INT64(pRaw, dataPos, consumerId);
}
SDB_SET_RESERVE(pRaw, dataPos, MND_CONSUMER_RESERVE_SIZE);
SDB_SET_DATALEN(pRaw, dataPos);
return pRaw;
}
static SSdbRow *mndCGroupActionDecode(SSdbRaw *pRaw) {
int8_t sver = 0;
if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
if (sver != MND_CONSUMER_VER_NUMBER) {
terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
mError("failed to decode cgroup since %s", terrstr());
return NULL;
}
// TODO: maximum size is not known
int32_t size = sizeof(SCGroupObj) + 128 * sizeof(int64_t);
SSdbRow *pRow = sdbAllocRow(size);
SCGroupObj *pCGroup = sdbGetRowObj(pRow);
if (pCGroup == NULL) return NULL;
int32_t dataPos = 0;
SDB_GET_BINARY(pRaw, pRow, dataPos, pCGroup->name, TSDB_TABLE_FNAME_LEN);
SDB_GET_INT64(pRaw, pRow, dataPos, &pCGroup->createTime);
SDB_GET_INT64(pRaw, pRow, dataPos, &pCGroup->updateTime);
SDB_GET_INT64(pRaw, pRow, dataPos, &pCGroup->uid);
/*SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->dbUid);*/
SDB_GET_INT32(pRaw, pRow, dataPos, &pCGroup->version);
int32_t sz;
SDB_GET_INT32(pRaw, pRow, dataPos, &sz);
// TODO: free list when failing
tdListInit(pCGroup->consumerIds, sizeof(int64_t));
for (int i = 0; i < sz; i++) {
int64_t consumerId;
SDB_GET_INT64(pRaw, pRow, dataPos, &consumerId);
tdListAppend(pCGroup->consumerIds, &consumerId);
}
SDB_GET_RESERVE(pRaw, pRow, dataPos, MND_CONSUMER_RESERVE_SIZE);
return pRow;
}
static SSdbRaw *mndConsumerActionEncode(SConsumerObj *pConsumer) { static SSdbRaw *mndConsumerActionEncode(SConsumerObj *pConsumer) {
int32_t size = sizeof(SConsumerObj) + MND_CONSUMER_RESERVE_SIZE; int32_t size = sizeof(SConsumerObj) + MND_CONSUMER_RESERVE_SIZE;
SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size); SSdbRaw *pRaw = sdbAllocRaw(SDB_CONSUMER, MND_CONSUMER_VER_NUMBER, size);
if (pRaw == NULL) return NULL; if (pRaw == NULL) return NULL;
int32_t dataPos = 0; int32_t dataPos = 0;
SDB_SET_BINARY(pRaw, dataPos, pConsumer->name, TSDB_TABLE_FNAME_LEN); SDB_SET_INT64(pRaw, dataPos, pConsumer->uid);
SDB_SET_BINARY(pRaw, dataPos, pConsumer->db, TSDB_DB_FNAME_LEN);
SDB_SET_INT64(pRaw, dataPos, pConsumer->createTime); SDB_SET_INT64(pRaw, dataPos, pConsumer->createTime);
SDB_SET_INT64(pRaw, dataPos, pConsumer->updateTime); SDB_SET_INT64(pRaw, dataPos, pConsumer->updateTime);
SDB_SET_INT64(pRaw, dataPos, pConsumer->uid);
/*SDB_SET_INT64(pRaw, dataPos, pConsumer->dbUid);*/ /*SDB_SET_INT64(pRaw, dataPos, pConsumer->dbUid);*/
SDB_SET_INT32(pRaw, dataPos, pConsumer->version); SDB_SET_INT32(pRaw, dataPos, pConsumer->version);
@ -102,11 +167,9 @@ static SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
if (pConsumer == NULL) return NULL; if (pConsumer == NULL) return NULL;
int32_t dataPos = 0; int32_t dataPos = 0;
SDB_GET_BINARY(pRaw, pRow, dataPos, pConsumer->name, TSDB_TABLE_FNAME_LEN); SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->uid);
SDB_GET_BINARY(pRaw, pRow, dataPos, pConsumer->db, TSDB_DB_FNAME_LEN);
SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->createTime); SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->createTime);
SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->updateTime); SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->updateTime);
SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->uid);
/*SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->dbUid);*/ /*SDB_GET_INT64(pRaw, pRow, dataPos, &pConsumer->dbUid);*/
SDB_GET_INT32(pRaw, pRow, dataPos, &pConsumer->version); SDB_GET_INT32(pRaw, pRow, dataPos, &pConsumer->version);
@ -116,17 +179,17 @@ static SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw) {
} }
static int32_t mndConsumerActionInsert(SSdb *pSdb, SConsumerObj *pConsumer) { static int32_t mndConsumerActionInsert(SSdb *pSdb, SConsumerObj *pConsumer) {
mTrace("consumer:%s, perform insert action", pConsumer->name); mTrace("consumer:%ld, perform insert action", pConsumer->uid);
return 0; return 0;
} }
static int32_t mndConsumerActionDelete(SSdb *pSdb, SConsumerObj *pConsumer) { static int32_t mndConsumerActionDelete(SSdb *pSdb, SConsumerObj *pConsumer) {
mTrace("consumer:%s, perform delete action", pConsumer->name); mTrace("consumer:%ld, perform delete action", pConsumer->uid);
return 0; return 0;
} }
static int32_t mndConsumerActionUpdate(SSdb *pSdb, SConsumerObj *pOldConsumer, SConsumerObj *pNewConsumer) { static int32_t mndConsumerActionUpdate(SSdb *pSdb, SConsumerObj *pOldConsumer, SConsumerObj *pNewConsumer) {
mTrace("consumer:%s, perform update action", pOldConsumer->name); mTrace("consumer:%ld, perform update action", pOldConsumer->uid);
atomic_exchange_32(&pOldConsumer->updateTime, pNewConsumer->updateTime); atomic_exchange_32(&pOldConsumer->updateTime, pNewConsumer->updateTime);
atomic_exchange_32(&pOldConsumer->version, pNewConsumer->version); atomic_exchange_32(&pOldConsumer->version, pNewConsumer->version);
@ -157,9 +220,34 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
char *msgStr = pMsg->rpcMsg.pCont; char *msgStr = pMsg->rpcMsg.pCont;
SCMSubscribeReq *pSubscribe; SCMSubscribeReq *pSubscribe;
tDeserializeSCMSubscribeReq(msgStr, pSubscribe); tDeserializeSCMSubscribeReq(msgStr, pSubscribe);
// add consumerGroupId -> list<consumerId> to sdb int topicNum = pSubscribe->topicNum;
// add consumerId -> list<consumer> to sdb int64_t consumerId = pSubscribe->consumerId;
// add consumer -> list<consumerId> to sdb char *consumerGroup = pSubscribe->consumerGroup;
// get consumer group and add client into it
SCGroupObj *pCGroup = sdbAcquire(pMnode->pSdb, SDB_CGROUP, consumerGroup);
if (pCGroup != NULL) {
// iterate the list until finding the consumer
// add consumer to cgroup list if not found
// put new record
}
SConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
if (pConsumer != NULL) {
//reset topic list
}
for (int i = 0; i < topicNum; i++) {
char *topicName = pSubscribe->topicName[i];
STopicObj *pTopic = mndAcquireTopic(pMnode, topicName);
//get
// consumer id
SList *list = pTopic->consumerIds;
// add the consumer if not in the list
//
SList* topicList = pConsumer->topics;
//add to topic
}
return 0; return 0;
} }
@ -255,9 +343,7 @@ static int32_t mndGetNumOfConsumers(SMnode *pMnode, char *dbName, int32_t *pNumO
pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer); pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
if (pIter == NULL) break; if (pIter == NULL) break;
if (strcmp(pConsumer->db, dbName) == 0) { numOfConsumers++;
numOfConsumers++;
}
sdbRelease(pSdb, pConsumer); sdbRelease(pSdb, pConsumer);
} }
@ -316,11 +402,11 @@ static int32_t mndGetConsumerMeta(SMnodeMsg *pMsg, SShowObj *pShow, STableMetaMs
return 0; return 0;
} }
static int32_t mndRetrieveConsumer(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) { static int32_t mndRetrieveCGroup(SMnodeMsg *pMsg, SShowObj *pShow, char *data, int32_t rows) {
SMnode *pMnode = pMsg->pMnode; SMnode *pMnode = pMsg->pMnode;
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
int32_t numOfRows = 0; int32_t numOfRows = 0;
SConsumerObj *pConsumer = NULL; SCGroupObj *pCGroup = NULL;
int32_t cols = 0; int32_t cols = 0;
char *pWrite; char *pWrite;
char prefix[64] = {0}; char prefix[64] = {0};
@ -330,36 +416,28 @@ static int32_t mndRetrieveConsumer(SMnodeMsg *pMsg, SShowObj *pShow, char *data,
int32_t prefixLen = (int32_t)strlen(prefix); int32_t prefixLen = (int32_t)strlen(prefix);
while (numOfRows < rows) { while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer); pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pCGroup);
if (pShow->pIter == NULL) break; if (pShow->pIter == NULL) break;
if (strncmp(pConsumer->name, prefix, prefixLen) != 0) { if (strncmp(pCGroup->name, prefix, prefixLen) != 0) {
sdbRelease(pSdb, pConsumer); sdbRelease(pSdb, pCGroup);
continue; continue;
} }
cols = 0; cols = 0;
char consumerName[TSDB_TABLE_NAME_LEN] = {0}; char consumerName[TSDB_TABLE_NAME_LEN] = {0};
tstrncpy(consumerName, pConsumer->name + prefixLen, TSDB_TABLE_NAME_LEN); tstrncpy(consumerName, pCGroup->name + prefixLen, TSDB_TABLE_NAME_LEN);
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
STR_TO_VARSTR(pWrite, consumerName); STR_TO_VARSTR(pWrite, consumerName);
cols++; cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows; pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int64_t *)pWrite = pConsumer->createTime; *(int64_t *)pWrite = pCGroup->createTime;
cols++; cols++;
/*pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;*/
/**(int32_t *)pWrite = pConsumer->numOfColumns;*/
/*cols++;*/
/*pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;*/
/**(int32_t *)pWrite = pConsumer->numOfTags;*/
/*cols++;*/
numOfRows++; numOfRows++;
sdbRelease(pSdb, pConsumer); sdbRelease(pSdb, pCGroup);
} }
pShow->numOfReads += numOfRows; pShow->numOfReads += numOfRows;