提交 31f8cf3f 编写于 作者: S slguan

Organize mgmtUtil.c file

上级 d97941aa
...@@ -30,10 +30,10 @@ extern "C" { ...@@ -30,10 +30,10 @@ extern "C" {
#include "tsdb.h" #include "tsdb.h"
#define UTIL_METER_IS_SUPERTABLE(metaInfo) \ #define UTIL_METER_IS_SUPERTABLE(metaInfo) \
(((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->meterType == TSDB_TABLE_TYPE_SUPER_TABLE)) (((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->tableType == TSDB_TABLE_TYPE_SUPER_TABLE))
#define UTIL_METER_IS_NOMRAL_METER(metaInfo) (!(UTIL_METER_IS_SUPERTABLE(metaInfo))) #define UTIL_METER_IS_NOMRAL_METER(metaInfo) (!(UTIL_METER_IS_SUPERTABLE(metaInfo)))
#define UTIL_METER_IS_CREATE_FROM_METRIC(metaInfo) \ #define UTIL_METER_IS_CREATE_FROM_METRIC(metaInfo) \
(((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->meterType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE)) (((metaInfo)->pMeterMeta != NULL) && ((metaInfo)->pMeterMeta->tableType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE))
#define TSDB_COL_IS_TAG(f) (((f)&TSDB_COL_TAG) != 0) #define TSDB_COL_IS_TAG(f) (((f)&TSDB_COL_TAG) != 0)
......
...@@ -54,7 +54,7 @@ typedef struct SMeterMetaInfo { ...@@ -54,7 +54,7 @@ typedef struct SMeterMetaInfo {
* 2. keep the vnode index for multi-vnode insertion * 2. keep the vnode index for multi-vnode insertion
*/ */
int32_t vnodeIndex; int32_t vnodeIndex;
char name[TSDB_METER_ID_LEN + 1]; // table(super table) name char name[TSDB_TABLE_ID_LEN + 1]; // table(super table) name
int16_t numOfTags; // total required tags in query, including groupby tags int16_t numOfTags; // total required tags in query, including groupby tags
int16_t tagColumnIndex[TSDB_MAX_TAGS]; // clause + tag projection int16_t tagColumnIndex[TSDB_MAX_TAGS]; // clause + tag projection
} SMeterMetaInfo; } SMeterMetaInfo;
...@@ -119,7 +119,7 @@ typedef struct SCond { ...@@ -119,7 +119,7 @@ typedef struct SCond {
} SCond; } SCond;
typedef struct SJoinNode { typedef struct SJoinNode {
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
uint64_t uid; uint64_t uid;
int16_t tagCol; int16_t tagCol;
} SJoinNode; } SJoinNode;
...@@ -154,7 +154,7 @@ typedef struct SParamInfo { ...@@ -154,7 +154,7 @@ typedef struct SParamInfo {
} SParamInfo; } SParamInfo;
typedef struct STableDataBlocks { typedef struct STableDataBlocks {
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
int8_t tsSource; // where does the UNIX timestamp come from, server or client int8_t tsSource; // where does the UNIX timestamp come from, server or client
bool ordered; // if current rows are ordered or not bool ordered; // if current rows are ordered or not
int64_t vgid; // virtual group id int64_t vgid; // virtual group id
...@@ -302,7 +302,7 @@ typedef struct _tsc_obj { ...@@ -302,7 +302,7 @@ typedef struct _tsc_obj {
char user[TSDB_USER_LEN]; char user[TSDB_USER_LEN];
char pass[TSDB_KEY_LEN]; char pass[TSDB_KEY_LEN];
char acctId[TSDB_DB_NAME_LEN]; char acctId[TSDB_DB_NAME_LEN];
char db[TSDB_METER_ID_LEN]; char db[TSDB_TABLE_ID_LEN];
char sversion[TSDB_VERSION_LEN]; char sversion[TSDB_VERSION_LEN];
char writeAuth : 1; char writeAuth : 1;
char superAuth : 1; char superAuth : 1;
......
...@@ -79,8 +79,8 @@ static int32_t getToStringLength(const char *pData, int32_t length, int32_t type ...@@ -79,8 +79,8 @@ static int32_t getToStringLength(const char *pData, int32_t length, int32_t type
static int32_t tscMaxLengthOfTagsFields(SSqlObj *pSql) { static int32_t tscMaxLengthOfTagsFields(SSqlObj *pSql) {
SMeterMeta *pMeta = tscGetMeterMetaInfo(&pSql->cmd, 0, 0)->pMeterMeta; SMeterMeta *pMeta = tscGetMeterMetaInfo(&pSql->cmd, 0, 0)->pMeterMeta;
if (pMeta->meterType == TSDB_TABLE_TYPE_SUPER_TABLE || pMeta->meterType == TSDB_TABLE_TYPE_NORMAL_TABLE || if (pMeta->tableType == TSDB_TABLE_TYPE_SUPER_TABLE || pMeta->tableType == TSDB_TABLE_TYPE_NORMAL_TABLE ||
pMeta->meterType == TSDB_TABLE_TYPE_STREAM_TABLE) { pMeta->tableType == TSDB_TABLE_TYPE_STREAM_TABLE) {
return 0; return 0;
} }
......
...@@ -776,7 +776,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) { ...@@ -776,7 +776,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
SMeterMetaInfo *pSTableMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, STABLE_INDEX); SMeterMetaInfo *pSTableMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, STABLE_INDEX);
setMeterID(pSTableMeterMetaInfo, &sToken, pSql); setMeterID(pSTableMeterMetaInfo, &sToken, pSql);
strncpy(pTag->name, pSTableMeterMetaInfo->name, TSDB_METER_ID_LEN); strncpy(pTag->name, pSTableMeterMetaInfo->name, TSDB_TABLE_ID_LEN);
code = tscGetMeterMeta(pSql, pSTableMeterMetaInfo); code = tscGetMeterMeta(pSql, pSTableMeterMetaInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
...@@ -950,7 +950,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) { ...@@ -950,7 +950,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
} }
int validateTableName(char *tblName, int len) { int validateTableName(char *tblName, int len) {
char buf[TSDB_METER_ID_LEN] = {0}; char buf[TSDB_TABLE_ID_LEN] = {0};
strncpy(buf, tblName, len); strncpy(buf, tblName, len);
SSQLToken token = {.n = len, .type = TK_ID, .z = buf}; SSQLToken token = {.n = len, .type = TK_ID, .z = buf};
...@@ -1544,7 +1544,7 @@ void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql) { ...@@ -1544,7 +1544,7 @@ void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql) {
continue; continue;
} }
strncpy(pMeterMetaInfo->name, pDataBlock->meterId, TSDB_METER_ID_LEN); strncpy(pMeterMetaInfo->name, pDataBlock->meterId, TSDB_TABLE_ID_LEN);
memset(pDataBlock->pData, 0, pDataBlock->nAllocSize); memset(pDataBlock->pData, 0, pDataBlock->nAllocSize);
int32_t ret = tscGetMeterMeta(pSql, pMeterMetaInfo); int32_t ret = tscGetMeterMeta(pSql, pMeterMetaInfo);
......
...@@ -1072,11 +1072,11 @@ int32_t setObjFullName(char* fullName, const char* account, SSQLToken* pDB, SSQL ...@@ -1072,11 +1072,11 @@ int32_t setObjFullName(char* fullName, const char* account, SSQLToken* pDB, SSQL
*xlen = totalLen; *xlen = totalLen;
} }
if (totalLen < TSDB_METER_ID_LEN) { if (totalLen < TSDB_TABLE_ID_LEN) {
fullName[totalLen] = 0; fullName[totalLen] = 0;
} }
return (totalLen <= TSDB_METER_ID_LEN) ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_SQL; return (totalLen <= TSDB_TABLE_ID_LEN) ? TSDB_CODE_SUCCESS : TSDB_CODE_INVALID_SQL;
} }
static void extractColumnNameFromString(tSQLExprItem* pItem) { static void extractColumnNameFromString(tSQLExprItem* pItem) {
...@@ -1901,7 +1901,7 @@ int32_t getMeterIndex(SSQLToken* pTableToken, SQueryInfo* pQueryInfo, SColumnInd ...@@ -1901,7 +1901,7 @@ int32_t getMeterIndex(SSQLToken* pTableToken, SQueryInfo* pQueryInfo, SColumnInd
} }
pIndex->tableIndex = COLUMN_INDEX_INITIAL_VAL; pIndex->tableIndex = COLUMN_INDEX_INITIAL_VAL;
char tableName[TSDB_METER_ID_LEN + 1] = {0}; char tableName[TSDB_TABLE_ID_LEN + 1] = {0};
for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) { for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, i); SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, i);
...@@ -3365,7 +3365,7 @@ static int32_t setTableCondForMetricQuery(SQueryInfo* pQueryInfo, const char* ac ...@@ -3365,7 +3365,7 @@ static int32_t setTableCondForMetricQuery(SQueryInfo* pQueryInfo, const char* ac
SStringBuilder sb1 = {0}; SStringBuilder sb1 = {0};
taosStringBuilderAppendStringLen(&sb1, QUERY_COND_REL_PREFIX_IN, QUERY_COND_REL_PREFIX_IN_LEN); taosStringBuilderAppendStringLen(&sb1, QUERY_COND_REL_PREFIX_IN, QUERY_COND_REL_PREFIX_IN_LEN);
char db[TSDB_METER_ID_LEN] = {0}; char db[TSDB_TABLE_ID_LEN] = {0};
// remove the duplicated input table names // remove the duplicated input table names
int32_t num = 0; int32_t num = 0;
...@@ -3389,7 +3389,7 @@ static int32_t setTableCondForMetricQuery(SQueryInfo* pQueryInfo, const char* ac ...@@ -3389,7 +3389,7 @@ static int32_t setTableCondForMetricQuery(SQueryInfo* pQueryInfo, const char* ac
taosStringBuilderAppendStringLen(&sb1, TBNAME_LIST_SEP, 1); taosStringBuilderAppendStringLen(&sb1, TBNAME_LIST_SEP, 1);
} }
char idBuf[TSDB_METER_ID_LEN + 1] = {0}; char idBuf[TSDB_TABLE_ID_LEN + 1] = {0};
int32_t xlen = strlen(segments[i]); int32_t xlen = strlen(segments[i]);
SSQLToken t = {.z = segments[i], .n = xlen, .type = TK_STRING}; SSQLToken t = {.z = segments[i], .n = xlen, .type = TK_STRING};
...@@ -5300,7 +5300,7 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) { ...@@ -5300,7 +5300,7 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) {
} }
// get meter meta from mnode // get meter meta from mnode
strncpy(pCreateTable->usingInfo.tagdata.name, pStableMeterMetaInfo->name, TSDB_METER_ID_LEN); strncpy(pCreateTable->usingInfo.tagdata.name, pStableMeterMetaInfo->name, TSDB_TABLE_ID_LEN);
tVariantList* pList = pInfo->pCreateTableInfo->usingInfo.pTagVals; tVariantList* pList = pInfo->pCreateTableInfo->usingInfo.pTagVals;
int32_t code = tscGetMeterMeta(pSql, pStableMeterMetaInfo); int32_t code = tscGetMeterMeta(pSql, pStableMeterMetaInfo);
......
...@@ -2685,7 +2685,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) { ...@@ -2685,7 +2685,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
// fill head info // fill head info
SMgmtHead *pMgmt = (SMgmtHead *)(pCmd->payload + tsRpcHeadSize); SMgmtHead *pMgmt = (SMgmtHead *)(pCmd->payload + tsRpcHeadSize);
memset(pMgmt->db, 0, TSDB_METER_ID_LEN); // server don't need the db memset(pMgmt->db, 0, TSDB_TABLE_ID_LEN); // server don't need the db
SMultiMeterInfoMsg *pInfoMsg = (SMultiMeterInfoMsg *)(pCmd->payload + tsRpcHeadSize + sizeof(SMgmtHead)); SMultiMeterInfoMsg *pInfoMsg = (SMultiMeterInfoMsg *)(pCmd->payload + tsRpcHeadSize + sizeof(SMgmtHead));
pInfoMsg->numOfMeters = htonl((int32_t)pCmd->count); pInfoMsg->numOfMeters = htonl((int32_t)pCmd->count);
...@@ -2709,7 +2709,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) { ...@@ -2709,7 +2709,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) { static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) {
const int32_t defaultSize = const int32_t defaultSize =
minMsgSize() + sizeof(SMetricMetaMsg) + sizeof(SMgmtHead) + sizeof(int16_t) * TSDB_MAX_TAGS; minMsgSize() + sizeof(SSuperTableMetaMsg) + sizeof(SMgmtHead) + sizeof(int16_t) * TSDB_MAX_TAGS;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
int32_t n = 0; int32_t n = 0;
...@@ -2722,7 +2722,7 @@ static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) { ...@@ -2722,7 +2722,7 @@ static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) {
tagLen += strlen(pQueryInfo->tagCond.tbnameCond.cond) * TSDB_NCHAR_SIZE; tagLen += strlen(pQueryInfo->tagCond.tbnameCond.cond) * TSDB_NCHAR_SIZE;
} }
int32_t joinCondLen = (TSDB_METER_ID_LEN + sizeof(int16_t)) * 2; int32_t joinCondLen = (TSDB_TABLE_ID_LEN + sizeof(int16_t)) * 2;
int32_t elemSize = sizeof(SMetricMetaElemMsg) * pQueryInfo->numOfTables; int32_t elemSize = sizeof(SMetricMetaElemMsg) * pQueryInfo->numOfTables;
int32_t len = tagLen + joinCondLen + elemSize + defaultSize; int32_t len = tagLen + joinCondLen + elemSize + defaultSize;
...@@ -2731,7 +2731,7 @@ static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) { ...@@ -2731,7 +2731,7 @@ static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) {
} }
int tscBuildMetricMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) { int tscBuildMetricMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SMetricMetaMsg *pMetaMsg; SSuperTableMetaMsg *pMetaMsg;
char * pMsg, *pStart; char * pMsg, *pStart;
int msgLen = 0; int msgLen = 0;
int tableIndex = 0; int tableIndex = 0;
...@@ -2757,25 +2757,25 @@ int tscBuildMetricMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) { ...@@ -2757,25 +2757,25 @@ int tscBuildMetricMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg += sizeof(SMgmtHead); pMsg += sizeof(SMgmtHead);
pMetaMsg = (SMetricMetaMsg *)pMsg; pMetaMsg = (SSuperTableMetaMsg *)pMsg;
pMetaMsg->numOfMeters = htonl(pQueryInfo->numOfTables); pMetaMsg->numOfMeters = htonl(pQueryInfo->numOfTables);
pMsg += sizeof(SMetricMetaMsg); pMsg += sizeof(SSuperTableMetaMsg);
int32_t offset = pMsg - (char *)pMetaMsg; int32_t offset = pMsg - (char *)pMetaMsg;
pMetaMsg->join = htonl(offset); pMetaMsg->join = htonl(offset);
// todo refactor // todo refactor
pMetaMsg->joinCondLen = htonl((TSDB_METER_ID_LEN + sizeof(int16_t)) * 2); pMetaMsg->joinCondLen = htonl((TSDB_TABLE_ID_LEN + sizeof(int16_t)) * 2);
memcpy(pMsg, pTagCond->joinInfo.left.meterId, TSDB_METER_ID_LEN); memcpy(pMsg, pTagCond->joinInfo.left.meterId, TSDB_TABLE_ID_LEN);
pMsg += TSDB_METER_ID_LEN; pMsg += TSDB_TABLE_ID_LEN;
*(int16_t *)pMsg = pTagCond->joinInfo.left.tagCol; *(int16_t *)pMsg = pTagCond->joinInfo.left.tagCol;
pMsg += sizeof(int16_t); pMsg += sizeof(int16_t);
memcpy(pMsg, pTagCond->joinInfo.right.meterId, TSDB_METER_ID_LEN); memcpy(pMsg, pTagCond->joinInfo.right.meterId, TSDB_TABLE_ID_LEN);
pMsg += TSDB_METER_ID_LEN; pMsg += TSDB_TABLE_ID_LEN;
*(int16_t *)pMsg = pTagCond->joinInfo.right.tagCol; *(int16_t *)pMsg = pTagCond->joinInfo.right.tagCol;
pMsg += sizeof(int16_t); pMsg += sizeof(int16_t);
...@@ -2991,7 +2991,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) { ...@@ -2991,7 +2991,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) {
int32_t tagLen = 0; int32_t tagLen = 0;
SSchema *pTagsSchema = tsGetTagSchema(pMeta); SSchema *pTagsSchema = tsGetTagSchema(pMeta);
if (pMeta->meterType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE) { if (pMeta->tableType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE) {
for (int32_t i = 0; i < pMeta->numOfTags; ++i) { for (int32_t i = 0; i < pMeta->numOfTags; ++i) {
tagLen += pTagsSchema[i].bytes; tagLen += pTagsSchema[i].bytes;
} }
...@@ -3106,7 +3106,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) { ...@@ -3106,7 +3106,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
int32_t tagLen = 0; int32_t tagLen = 0;
SSchema *pTagsSchema = tsGetTagSchema(pMeta); SSchema *pTagsSchema = tsGetTagSchema(pMeta);
if (pMeta->meterType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE) { if (pMeta->tableType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE) {
for (int32_t j = 0; j < pMeta->numOfTags; ++j) { for (int32_t j = 0; j < pMeta->numOfTags; ++j) {
tagLen += pTagsSchema[j].bytes; tagLen += pTagsSchema[j].bytes;
} }
...@@ -3304,7 +3304,7 @@ int tscProcessShowRsp(SSqlObj *pSql) { ...@@ -3304,7 +3304,7 @@ int tscProcessShowRsp(SSqlObj *pSql) {
} }
int tscProcessConnectRsp(SSqlObj *pSql) { int tscProcessConnectRsp(SSqlObj *pSql) {
char temp[TSDB_METER_ID_LEN * 2]; char temp[TSDB_TABLE_ID_LEN * 2];
SConnectRsp *pConnect; SConnectRsp *pConnect;
STscObj *pObj = pSql->pTscObj; STscObj *pObj = pSql->pTscObj;
......
...@@ -1047,7 +1047,7 @@ static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t t ...@@ -1047,7 +1047,7 @@ static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t t
} }
char *nextStr; char *nextStr;
char tblName[TSDB_METER_ID_LEN]; char tblName[TSDB_TABLE_ID_LEN];
int payloadLen = 0; int payloadLen = 0;
char *pMsg = pCmd->payload; char *pMsg = pCmd->payload;
while (1) { while (1) {
......
...@@ -657,7 +657,7 @@ int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOff ...@@ -657,7 +657,7 @@ int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOff
dataBuf->size = startOffset; dataBuf->size = startOffset;
dataBuf->tsSource = -1; dataBuf->tsSource = -1;
strncpy(dataBuf->meterId, name, TSDB_METER_ID_LEN); strncpy(dataBuf->meterId, name, TSDB_TABLE_ID_LEN);
/* /*
* The metermeta may be released since the metermeta cache are completed clean by other thread * The metermeta may be released since the metermeta cache are completed clean by other thread
...@@ -1810,7 +1810,7 @@ SMeterMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, SM ...@@ -1810,7 +1810,7 @@ SMeterMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, SM
assert(pMeterMetaInfo != NULL); assert(pMeterMetaInfo != NULL);
if (name != NULL) { if (name != NULL) {
assert(strlen(name) <= TSDB_METER_ID_LEN); assert(strlen(name) <= TSDB_TABLE_ID_LEN);
strcpy(pMeterMetaInfo->name, name); strcpy(pMeterMetaInfo->name, name);
} }
......
...@@ -312,7 +312,7 @@ int vnodeProcessCreateMeterMsg(char *pMsg, int msgLen) { ...@@ -312,7 +312,7 @@ int vnodeProcessCreateMeterMsg(char *pMsg, int msgLen) {
pObj->vnode = pCreate->vnode; pObj->vnode = pCreate->vnode;
pObj->sid = pCreate->sid; pObj->sid = pCreate->sid;
pObj->uid = pCreate->uid; pObj->uid = pCreate->uid;
memcpy(pObj->meterId, pCreate->meterId, TSDB_METER_ID_LEN); memcpy(pObj->meterId, pCreate->meterId, TSDB_TABLE_ID_LEN);
pObj->numOfColumns = pCreate->numOfColumns; pObj->numOfColumns = pCreate->numOfColumns;
pObj->timeStamp = pCreate->timeStamp; pObj->timeStamp = pCreate->timeStamp;
pObj->sversion = htonl(pCreate->sversion); pObj->sversion = htonl(pCreate->sversion);
...@@ -377,7 +377,7 @@ int vnodeProcessRemoveMeterRequest(char *pMsg, int msgLen, SMgmtObj *pMgmtObj) { ...@@ -377,7 +377,7 @@ int vnodeProcessRemoveMeterRequest(char *pMsg, int msgLen, SMgmtObj *pMgmtObj) {
pObj = vnodeList[pRemove->vnode].meterList[pRemove->sid]; pObj = vnodeList[pRemove->vnode].meterList[pRemove->sid];
if (pObj == NULL) goto _remove_over; if (pObj == NULL) goto _remove_over;
if (memcmp(pObj->meterId, pRemove->meterId, TSDB_METER_ID_LEN) != 0) { if (memcmp(pObj->meterId, pRemove->meterId, TSDB_TABLE_ID_LEN) != 0) {
dWarn("vid:%d sid:%d id:%s, remove ID:%s, meter ID not matched", pObj->vnode, pObj->sid, pObj->meterId, dWarn("vid:%d sid:%d id:%s, remove ID:%s, meter ID not matched", pObj->vnode, pObj->sid, pObj->meterId,
pRemove->meterId); pRemove->meterId);
goto _remove_over; goto _remove_over;
......
...@@ -95,7 +95,7 @@ typedef struct { ...@@ -95,7 +95,7 @@ typedef struct {
} SMeterGid; } SMeterGid;
typedef struct _tab_obj { typedef struct _tab_obj {
char meterId[TSDB_METER_ID_LEN + 1]; char meterId[TSDB_TABLE_ID_LEN + 1];
uint64_t uid; uint64_t uid;
SMeterGid gid; SMeterGid gid;
...@@ -106,7 +106,7 @@ typedef struct _tab_obj { ...@@ -106,7 +106,7 @@ typedef struct _tab_obj {
int32_t numOfColumns; int32_t numOfColumns;
int32_t schemaSize; int32_t schemaSize;
short nextColId; short nextColId;
char meterType : 4; char tableType : 4;
char status : 3; char status : 3;
char isDirty : 1; // if the table change tag column 1 value char isDirty : 1; // if the table change tag column 1 value
char reserved[15]; char reserved[15];
...@@ -116,7 +116,7 @@ typedef struct _tab_obj { ...@@ -116,7 +116,7 @@ typedef struct _tab_obj {
tSkipList * pSkipList; tSkipList * pSkipList;
struct _tab_obj *pHead; // for metric, a link list for all meters created struct _tab_obj *pHead; // for metric, a link list for all meters created
// according to this metric // according to this metric
char *pTagData; // TSDB_METER_ID_LEN(metric_name)+ char *pTagData; // TSDB_TABLE_ID_LEN(metric_name)+
// tags_value1/tags_value2/tags_value3 // tags_value1/tags_value2/tags_value3
struct _tab_obj *prev, *next; struct _tab_obj *prev, *next;
char * pSql; // pointer to SQL, for SC, null-terminated string char * pSql; // pointer to SQL, for SC, null-terminated string
...@@ -262,8 +262,8 @@ extern SDnodeObj dnodeObj; ...@@ -262,8 +262,8 @@ extern SDnodeObj dnodeObj;
// dnodeInt API // dnodeInt API
int mgmtInitDnodeInt(); int mgmtInitDnodeInt();
void mgmtCleanUpDnodeInt(); void mgmtCleanUpDnodeInt();
int mgmtSendCreateMsgToVgroup(STabObj *pMeter, SVgObj *pVgroup); int mgmtSendCreateMsgToVgroup(STabObj *pTable, SVgObj *pVgroup);
int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup); int mgmtSendRemoveMeterMsgToDnode(STabObj *pTable, SVgObj *pVgroup);
int mgmtSendVPeersMsg(SVgObj *pVgroup); int mgmtSendVPeersMsg(SVgObj *pVgroup);
int mgmtSendFreeVnodeMsg(SVgObj *pVgroup); int mgmtSendFreeVnodeMsg(SVgObj *pVgroup);
int mgmtSendOneFreeVnodeMsg(SVnodeGid *pVnodeGid); int mgmtSendOneFreeVnodeMsg(SVnodeGid *pVnodeGid);
...@@ -284,8 +284,8 @@ int mgmtRetrieveUsers(SShowObj *pShow, char *data, int rows, SConnObj *pConn); ...@@ -284,8 +284,8 @@ int mgmtRetrieveUsers(SShowObj *pShow, char *data, int rows, SConnObj *pConn);
void mgmtCleanUpUsers(); void mgmtCleanUpUsers();
// metric API // metric API
int mgmtAddMeterIntoMetric(STabObj *pMetric, STabObj *pMeter); int mgmtAddMeterIntoMetric(STabObj *pMetric, STabObj *pTable);
int mgmtRemoveMeterFromMetric(STabObj *pMetric, STabObj *pMeter); int mgmtRemoveMeterFromMetric(STabObj *pMetric, STabObj *pTable);
int mgmtGetMetricMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn); int mgmtGetMetricMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn);
int mgmtRetrieveMetrics(SShowObj *pShow, char *data, int rows, SConnObj *pConn); int mgmtRetrieveMetrics(SShowObj *pShow, char *data, int rows, SConnObj *pConn);
...@@ -326,14 +326,14 @@ void mgmtCleanUpVgroups(); ...@@ -326,14 +326,14 @@ void mgmtCleanUpVgroups();
int mgmtInitMeters(); int mgmtInitMeters();
STabObj *mgmtGetTable(char *meterId); STabObj *mgmtGetTable(char *meterId);
STabObj *mgmtGetTableInfo(char *src, char *tags[]); STabObj *mgmtGetTableInfo(char *src, char *tags[]);
int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SMetricMetaMsg *pInfo); int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SSuperTableMetaMsg *pInfo);
int mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate); int mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate);
int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore); int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore);
int mgmtAlterMeter(SDbObj *pDb, SAlterTableMsg *pAlter); int mgmtAlterMeter(SDbObj *pDb, SAlterTableMsg *pAlter);
int mgmtGetTableMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn); int mgmtGetTableMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn);
int mgmtRetrieveMeters(SShowObj *pShow, char *data, int rows, SConnObj *pConn); int mgmtRetrieveMeters(SShowObj *pShow, char *data, int rows, SConnObj *pConn);
void mgmtCleanUpMeters(); void mgmtCleanUpMeters();
SSchema *mgmtGetTableSchema(STabObj *pMeter); // get schema for a meter SSchema *mgmtGetTableSchema(STabObj *pTable); // get schema for a meter
// dnode API // dnode API
......
...@@ -301,7 +301,7 @@ typedef struct { ...@@ -301,7 +301,7 @@ typedef struct {
uint64_t uid; uint64_t uid;
char spi; char spi;
char encrypt; char encrypt;
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
char secret[TSDB_KEY_LEN]; char secret[TSDB_KEY_LEN];
char cipheringKey[TSDB_KEY_LEN]; char cipheringKey[TSDB_KEY_LEN];
uint64_t timeStamp; uint64_t timeStamp;
...@@ -314,7 +314,7 @@ typedef struct { ...@@ -314,7 +314,7 @@ typedef struct {
} SCreateMsg; } SCreateMsg;
typedef struct { typedef struct {
char db[TSDB_METER_ID_LEN]; char db[TSDB_TABLE_ID_LEN];
uint8_t ignoreNotExists; uint8_t ignoreNotExists;
} SDropDbMsg, SUseDbMsg; } SDropDbMsg, SUseDbMsg;
...@@ -327,7 +327,7 @@ typedef struct { ...@@ -327,7 +327,7 @@ typedef struct {
} SShowTableMsg; } SShowTableMsg;
typedef struct { typedef struct {
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
char igExists; char igExists;
short numOfTags; short numOfTags;
...@@ -341,12 +341,12 @@ typedef struct { ...@@ -341,12 +341,12 @@ typedef struct {
} SCreateTableMsg; } SCreateTableMsg;
typedef struct { typedef struct {
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
char igNotExists; char igNotExists;
} SDropTableMsg; } SDropTableMsg;
typedef struct { typedef struct {
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
short type; /* operation type */ short type; /* operation type */
char tagVal[TSDB_MAX_BYTES_PER_ROW]; char tagVal[TSDB_MAX_BYTES_PER_ROW];
short numOfCols; /* number of schema */ short numOfCols; /* number of schema */
...@@ -355,7 +355,7 @@ typedef struct { ...@@ -355,7 +355,7 @@ typedef struct {
typedef struct { typedef struct {
char clientVersion[TSDB_VERSION_LEN]; char clientVersion[TSDB_VERSION_LEN];
char db[TSDB_METER_ID_LEN]; char db[TSDB_TABLE_ID_LEN];
} SConnectMsg; } SConnectMsg;
typedef struct { typedef struct {
...@@ -386,7 +386,7 @@ typedef struct { ...@@ -386,7 +386,7 @@ typedef struct {
} SCreateUserMsg, SAlterUserMsg; } SCreateUserMsg, SAlterUserMsg;
typedef struct { typedef struct {
char db[TSDB_METER_ID_LEN]; char db[TSDB_TABLE_ID_LEN];
} SMgmtHead; } SMgmtHead;
typedef struct { typedef struct {
...@@ -400,7 +400,7 @@ typedef struct { ...@@ -400,7 +400,7 @@ typedef struct {
short vnode; short vnode;
int32_t sid; int32_t sid;
uint64_t uid; uint64_t uid;
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
} SRemoveMeterMsg; } SRemoveMeterMsg;
typedef struct { typedef struct {
...@@ -604,7 +604,7 @@ typedef struct { ...@@ -604,7 +604,7 @@ typedef struct {
* the message is too large, so it may will overwrite the cfg information in meterobj.v* * the message is too large, so it may will overwrite the cfg information in meterobj.v*
* recover to origin codes * recover to origin codes
*/ */
//char db[TSDB_METER_ID_LEN+2]; // 8bytes align //char db[TSDB_TABLE_ID_LEN+2]; // 8bytes align
char db[TSDB_DB_NAME_LEN]; char db[TSDB_DB_NAME_LEN];
uint32_t vgId; uint32_t vgId;
int32_t maxSessions; int32_t maxSessions;
...@@ -695,7 +695,7 @@ typedef struct { ...@@ -695,7 +695,7 @@ typedef struct {
} SVPeersMsg; } SVPeersMsg;
typedef struct { typedef struct {
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
short createFlag; short createFlag;
char tags[]; char tags[];
} SMeterInfoMsg; } SMeterInfoMsg;
...@@ -708,7 +708,7 @@ typedef struct { ...@@ -708,7 +708,7 @@ typedef struct {
typedef struct { typedef struct {
int16_t elemLen; int16_t elemLen;
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
int16_t orderIndex; int16_t orderIndex;
int16_t orderType; // used in group by xx order by xxx int16_t orderType; // used in group by xx order by xxx
...@@ -732,7 +732,7 @@ typedef struct { ...@@ -732,7 +732,7 @@ typedef struct {
int32_t join; int32_t join;
int32_t joinCondLen; // for join condition int32_t joinCondLen; // for join condition
int32_t metaElem[TSDB_MAX_JOIN_TABLE_NUM]; int32_t metaElem[TSDB_MAX_JOIN_TABLE_NUM];
} SMetricMetaMsg; } SSuperTableMetaMsg;
typedef struct { typedef struct {
SVPeerDesc vpeerDesc[TSDB_VNODES_SUPPORT]; SVPeerDesc vpeerDesc[TSDB_VNODES_SUPPORT];
...@@ -751,7 +751,7 @@ typedef struct { ...@@ -751,7 +751,7 @@ typedef struct {
typedef struct SMeterMeta { typedef struct SMeterMeta {
uint8_t numOfTags : 6; uint8_t numOfTags : 6;
uint8_t precision : 2; uint8_t precision : 2;
uint8_t meterType : 4; uint8_t tableType : 4;
uint8_t index : 4; // used locally uint8_t index : 4; // used locally
int16_t numOfColumns; int16_t numOfColumns;
...@@ -767,12 +767,12 @@ typedef struct SMeterMeta { ...@@ -767,12 +767,12 @@ typedef struct SMeterMeta {
} SMeterMeta; } SMeterMeta;
typedef struct SMultiMeterMeta { typedef struct SMultiMeterMeta {
char meterId[TSDB_METER_ID_LEN]; // note: This field must be at the front char meterId[TSDB_TABLE_ID_LEN]; // note: This field must be at the front
SMeterMeta meta; SMeterMeta meta;
} SMultiMeterMeta; } SMultiMeterMeta;
typedef struct { typedef struct {
char name[TSDB_METER_ID_LEN]; char name[TSDB_TABLE_ID_LEN];
char data[TSDB_MAX_TAGS_LEN]; char data[TSDB_MAX_TAGS_LEN];
} STagData; } STagData;
......
...@@ -86,7 +86,7 @@ extern "C" { ...@@ -86,7 +86,7 @@ extern "C" {
#define TS_PATH_DELIMITER_LEN 1 #define TS_PATH_DELIMITER_LEN 1
#define TSDB_METER_ID_LEN_MARGIN 10 #define TSDB_METER_ID_LEN_MARGIN 10
#define TSDB_METER_ID_LEN (TSDB_DB_NAME_LEN+TSDB_METER_NAME_LEN+2*TS_PATH_DELIMITER_LEN+TSDB_USERID_LEN+TSDB_METER_ID_LEN_MARGIN) //TSDB_DB_NAME_LEN+TSDB_METER_NAME_LEN+2*strlen(TS_PATH_DELIMITER)+strlen(USERID) #define TSDB_TABLE_ID_LEN (TSDB_DB_NAME_LEN+TSDB_METER_NAME_LEN+2*TS_PATH_DELIMITER_LEN+TSDB_USERID_LEN+TSDB_METER_ID_LEN_MARGIN) //TSDB_DB_NAME_LEN+TSDB_METER_NAME_LEN+2*strlen(TS_PATH_DELIMITER)+strlen(USERID)
#define TSDB_UNI_LEN 24 #define TSDB_UNI_LEN 24
#define TSDB_USER_LEN TSDB_UNI_LEN #define TSDB_USER_LEN TSDB_UNI_LEN
#define TSDB_ACCT_LEN TSDB_UNI_LEN #define TSDB_ACCT_LEN TSDB_UNI_LEN
......
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* 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 <http://www.gnu.org/licenses/>.
*/
#ifndef TBASE_MNODE_SUPER_TABLE_QUERY_H
#define TBASE_MNODE_SUPER_TABLE_QUERY_H
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include "mnode.h"
int32_t mgmtRetrieveMetersFromSuperTable(SSuperTableMetaMsg* pInfo, int32_t tableIndex, tQueryResultset* pRes);
int32_t mgmtDoJoin(SSuperTableMetaMsg* pSuperTableMetaMsg, tQueryResultset* pRes);
void mgmtReorganizeMetersInMetricMeta(SSuperTableMetaMsg* pInfo, int32_t index, tQueryResultset* pRes);
#endif
...@@ -11,28 +11,16 @@ ...@@ -11,28 +11,16 @@
* *
* You should have received a copy of the GNU Affero General Public License * You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/#include <stdio.h> */
#include <stdlib.h>
#include <stdint.h>
#include "tast.h"
#ifndef TBASE_MNODE_UTIL_H #ifndef TBASE_MNODE_TABLE_H
#define TBASE_MNODE_UTIL_H #define TBASE_MNODE_TABLE_H
bool mgmtTableCreateFromSuperTable(STabObj *pTableObj); #include <stdio.h>
bool mgmtIsSuperTable(STabObj *pTableObj); #include <stdlib.h>
bool mgmtIsNormalTable(STabObj *pTableObj); #include <stdint.h>
#include "mnode.h"
typedef struct SSyntaxTreeFilterSupporter {
SSchema* pTagSchema;
int32_t numOfTags;
int32_t optr;
} SSyntaxTreeFilterSupporter;
char* mgmtTableGetTag(STabObj* pMeter, int32_t col, SSchema* pTagColSchema); int32_t mgmtFindTagCol(STabObj * pTable, const char * tagName);
int32_t mgmtGetTagsLength(STabObj* pMetric, int32_t col);
bool mgmtCheckIsMonitorDB(char *db, char *monitordb);
int32_t mgmtCheckDBParams(SCreateDbMsg *pCreate);
#endif #endif
...@@ -11,37 +11,22 @@ ...@@ -11,37 +11,22 @@
* *
* You should have received a copy of the GNU Affero General Public License * You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/#include <stdio.h> */
#include <stdlib.h>
#include <stdint.h>
#include "tast.h"
#ifndef TBASE_MNODE_UTIL_H #ifndef TBASE_MNODE_UTIL_H
#define TBASE_MNODE_UTIL_H #define TBASE_MNODE_UTIL_H
bool mgmtTableCreateFromSuperTable(STabObj *pTableObj); #include <stdio.h>
bool mgmtIsSuperTable(STabObj *pTableObj); #include <stdlib.h>
bool mgmtIsNormalTable(STabObj *pTableObj); #include <stdint.h>
#include "mnode.h"
typedef struct SSyntaxTreeFilterSupporter {
SSchema* pTagSchema;
int32_t numOfTags;
int32_t optr;
} SSyntaxTreeFilterSupporter;
char* mgmtTableGetTag(STabObj* pMeter, int32_t col, SSchema* pTagColSchema); bool mgmtTableCreateFromSuperTable(STabObj *pTableObj);
int32_t mgmtGetTagsLength(STabObj* pMetric, int32_t col); bool mgmtIsSuperTable(STabObj *pTableObj);
bool mgmtIsNormalTable(STabObj *pTableObj);
char* mgmtTableGetTag(STabObj* pTable, int32_t col, SSchema* pTagColSchema);
int32_t mgmtGetTagsLength(STabObj* pSuperTable, int32_t col);
bool mgmtCheckIsMonitorDB(char *db, char *monitordb); bool mgmtCheckIsMonitorDB(char *db, char *monitordb);
int32_t mgmtCheckDBParams(SCreateDbMsg *pCreate); int32_t mgmtCheckDBParams(SCreateDbMsg *pCreate);
int32_t mgmtFindTagCol(STabObj * pMetric, const char * tagName);
int32_t mgmtRetrieveMetersFromMetric(SMetricMetaMsg* pInfo, int32_t tableIndex, tQueryResultset* pRes);
int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes);
void mgmtReorganizeMetersInMetricMeta(SMetricMetaMsg* pInfo, int32_t index, tQueryResultset* pRes);
bool tSkipListNodeFilterCallback(const void *pNode, void *param);
#endif #endif
...@@ -20,7 +20,7 @@ ...@@ -20,7 +20,7 @@
#include "tschemautil.h" #include "tschemautil.h"
typedef struct { typedef struct {
char user[TSDB_METER_ID_LEN]; char user[TSDB_TABLE_ID_LEN];
uint64_t stime; uint64_t stime;
uint32_t ip; uint32_t ip;
uint16_t port; uint16_t port;
......
...@@ -116,7 +116,7 @@ int mgmtInitDbs() { ...@@ -116,7 +116,7 @@ int mgmtInitDbs() {
SDbObj *mgmtGetDb(char *db) { return (SDbObj *)sdbGetRow(dbSdb, db); } SDbObj *mgmtGetDb(char *db) { return (SDbObj *)sdbGetRow(dbSdb, db); }
SDbObj *mgmtGetDbByMeterId(char *meterId) { SDbObj *mgmtGetDbByMeterId(char *meterId) {
char db[TSDB_METER_ID_LEN], *pos; char db[TSDB_TABLE_ID_LEN], *pos;
pos = strstr(meterId, TS_PATH_DELIMITER); pos = strstr(meterId, TS_PATH_DELIMITER);
pos = strstr(pos + 1, TS_PATH_DELIMITER); pos = strstr(pos + 1, TS_PATH_DELIMITER);
......
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
void mgmtProcessMsgFromDnode(char *content, int msgLen, int msgType, SDnodeObj *pObj); void mgmtProcessMsgFromDnode(char *content, int msgLen, int msgType, SDnodeObj *pObj);
int mgmtSendVPeersMsg(SVgObj *pVgroup); int mgmtSendVPeersMsg(SVgObj *pVgroup);
char *mgmtBuildVpeersIe(char *pMsg, SVgObj *pVgroup, int vnode); char *mgmtBuildVpeersIe(char *pMsg, SVgObj *pVgroup, int vnode);
char *mgmtBuildCreateMeterIe(STabObj *pMeter, char *pMsg, int vnode); char *mgmtBuildCreateMeterIe(STabObj *pTable, char *pMsg, int vnode);
/* /*
* functions for communicate between dnode and mnode * functions for communicate between dnode and mnode
...@@ -39,7 +39,7 @@ int taosSendMsgToDnode(SDnodeObj *pObj, char *msg, int msgLen); ...@@ -39,7 +39,7 @@ int taosSendMsgToDnode(SDnodeObj *pObj, char *msg, int msgLen);
int mgmtProcessMeterCfgMsg(char *cont, int contLen, SDnodeObj *pObj) { int mgmtProcessMeterCfgMsg(char *cont, int contLen, SDnodeObj *pObj) {
char * pMsg, *pStart; char * pMsg, *pStart;
int msgLen = 0; int msgLen = 0;
STabObj * pMeter = NULL; STabObj * pTable = NULL;
SMeterCfgMsg *pCfg = (SMeterCfgMsg *)cont; SMeterCfgMsg *pCfg = (SMeterCfgMsg *)cont;
SVgObj * pVgroup; SVgObj * pVgroup;
...@@ -63,13 +63,13 @@ int mgmtProcessMeterCfgMsg(char *cont, int contLen, SDnodeObj *pObj) { ...@@ -63,13 +63,13 @@ int mgmtProcessMeterCfgMsg(char *cont, int contLen, SDnodeObj *pObj) {
int vgId = pObj->vload[vnode].vgId; int vgId = pObj->vload[vnode].vgId;
pVgroup = mgmtGetVgroup(vgId); pVgroup = mgmtGetVgroup(vgId);
if (pVgroup) pMeter = pVgroup->meterList[sid]; if (pVgroup) pTable = pVgroup->meterList[sid];
} }
if (pMeter) { if (pTable) {
*pMsg = 0; // code *pMsg = 0; // code
pMsg++; pMsg++;
pMsg = mgmtBuildCreateMeterIe(pMeter, pMsg, vnode); pMsg = mgmtBuildCreateMeterIe(pTable, pMsg, vnode);
} else { } else {
mTrace("dnode:%s, vnode:%d sid:%d, meter not there", taosIpStr(pObj->privateIp), vnode, sid); mTrace("dnode:%s, vnode:%d sid:%d, meter not there", taosIpStr(pObj->privateIp), vnode, sid);
*pMsg = TSDB_CODE_INVALID_METER_ID; *pMsg = TSDB_CODE_INVALID_METER_ID;
...@@ -187,48 +187,48 @@ void mgmtProcessMsgFromDnode(char *content, int msgLen, int msgType, SDnodeObj * ...@@ -187,48 +187,48 @@ void mgmtProcessMsgFromDnode(char *content, int msgLen, int msgType, SDnodeObj *
} }
} }
char *mgmtBuildCreateMeterIe(STabObj *pMeter, char *pMsg, int vnode) { char *mgmtBuildCreateMeterIe(STabObj *pTable, char *pMsg, int vnode) {
SCreateMsg *pCreateMeter; SCreateMsg *pCreateMeter;
pCreateMeter = (SCreateMsg *)pMsg; pCreateMeter = (SCreateMsg *)pMsg;
pCreateMeter->vnode = htons(vnode); pCreateMeter->vnode = htons(vnode);
pCreateMeter->sid = htonl(pMeter->gid.sid); pCreateMeter->sid = htonl(pTable->gid.sid);
pCreateMeter->uid = pMeter->uid; pCreateMeter->uid = pTable->uid;
memcpy(pCreateMeter->meterId, pMeter->meterId, TSDB_METER_ID_LEN); memcpy(pCreateMeter->meterId, pTable->meterId, TSDB_TABLE_ID_LEN);
// pCreateMeter->lastCreate = htobe64(pVgroup->lastCreate); // pCreateMeter->lastCreate = htobe64(pVgroup->lastCreate);
pCreateMeter->timeStamp = htobe64(pMeter->createdTime); pCreateMeter->timeStamp = htobe64(pTable->createdTime);
/* /*
pCreateMeter->spi = pSec->spi; pCreateMeter->spi = pSec->spi;
pCreateMeter->encrypt = pSec->encrypt; pCreateMeter->encrypt = pSec->encrypt;
memcpy(pCreateMeter->cipheringKey, pSec->cipheringKey, TSDB_KEY_LEN); memcpy(pCreateMeter->cipheringKey, pSec->cipheringKey, TSDB_KEY_LEN);
memcpy(pCreateMeter->secret, pSec->secret, TSDB_KEY_LEN); memcpy(pCreateMeter->secret, pSec->secret, TSDB_KEY_LEN);
*/ */
pCreateMeter->sversion = htonl(pMeter->sversion); pCreateMeter->sversion = htonl(pTable->sversion);
pCreateMeter->numOfColumns = htons(pMeter->numOfColumns); pCreateMeter->numOfColumns = htons(pTable->numOfColumns);
SSchema *pSchema = mgmtGetTableSchema(pMeter); SSchema *pSchema = mgmtGetTableSchema(pTable);
for (int i = 0; i < pMeter->numOfColumns; ++i) { for (int i = 0; i < pTable->numOfColumns; ++i) {
pCreateMeter->schema[i].type = pSchema[i].type; pCreateMeter->schema[i].type = pSchema[i].type;
/* strcpy(pCreateMeter->schema[i].name, pSchema[i].name); */ /* strcpy(pCreateMeter->schema[i].name, pSchema[i].name); */
pCreateMeter->schema[i].bytes = htons(pSchema[i].bytes); pCreateMeter->schema[i].bytes = htons(pSchema[i].bytes);
pCreateMeter->schema[i].colId = htons(pSchema[i].colId); pCreateMeter->schema[i].colId = htons(pSchema[i].colId);
} }
pMsg = ((char *)(pCreateMeter->schema)) + pMeter->numOfColumns * sizeof(SMColumn); pMsg = ((char *)(pCreateMeter->schema)) + pTable->numOfColumns * sizeof(SMColumn);
pCreateMeter->sqlLen = 0; pCreateMeter->sqlLen = 0;
if (pMeter->pSql) { if (pTable->pSql) {
int len = strlen(pMeter->pSql) + 1; int len = strlen(pTable->pSql) + 1;
pCreateMeter->sqlLen = htons(len); pCreateMeter->sqlLen = htons(len);
strcpy(pMsg, pMeter->pSql); strcpy(pMsg, pTable->pSql);
pMsg += len; pMsg += len;
} }
return pMsg; return pMsg;
} }
int mgmtSendCreateMsgToVgroup(STabObj *pMeter, SVgObj *pVgroup) { int mgmtSendCreateMsgToVgroup(STabObj *pTable, SVgObj *pVgroup) {
char * pMsg, *pStart; char * pMsg, *pStart;
int i, msgLen = 0; int i, msgLen = 0;
SDnodeObj *pObj; SDnodeObj *pObj;
...@@ -244,7 +244,7 @@ int mgmtSendCreateMsgToVgroup(STabObj *pMeter, SVgObj *pVgroup) { ...@@ -244,7 +244,7 @@ int mgmtSendCreateMsgToVgroup(STabObj *pMeter, SVgObj *pVgroup) {
pStart = taosBuildReqMsgToDnodeWithSize(pObj, TSDB_MSG_TYPE_CREATE, 64000); pStart = taosBuildReqMsgToDnodeWithSize(pObj, TSDB_MSG_TYPE_CREATE, 64000);
if (pStart == NULL) continue; if (pStart == NULL) continue;
pMsg = mgmtBuildCreateMeterIe(pMeter, pStart, pVgroup->vnodeGid[i].vnode); pMsg = mgmtBuildCreateMeterIe(pTable, pStart, pVgroup->vnodeGid[i].vnode);
msgLen = pMsg - pStart; msgLen = pMsg - pStart;
taosSendMsgToDnode(pObj, pStart, msgLen); taosSendMsgToDnode(pObj, pStart, msgLen);
...@@ -255,7 +255,7 @@ int mgmtSendCreateMsgToVgroup(STabObj *pMeter, SVgObj *pVgroup) { ...@@ -255,7 +255,7 @@ int mgmtSendCreateMsgToVgroup(STabObj *pMeter, SVgObj *pVgroup) {
return 0; return 0;
} }
int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) { int mgmtSendRemoveMeterMsgToDnode(STabObj *pTable, SVgObj *pVgroup) {
SRemoveMeterMsg *pRemove; SRemoveMeterMsg *pRemove;
char * pMsg, *pStart; char * pMsg, *pStart;
int i, msgLen = 0; int i, msgLen = 0;
...@@ -277,8 +277,8 @@ int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) { ...@@ -277,8 +277,8 @@ int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) {
pRemove = (SRemoveMeterMsg *)pMsg; pRemove = (SRemoveMeterMsg *)pMsg;
pRemove->vnode = htons(pVgroup->vnodeGid[i].vnode); pRemove->vnode = htons(pVgroup->vnodeGid[i].vnode);
pRemove->sid = htonl(pMeter->gid.sid); pRemove->sid = htonl(pTable->gid.sid);
memcpy(pRemove->meterId, pMeter->meterId, TSDB_METER_ID_LEN); memcpy(pRemove->meterId, pTable->meterId, TSDB_TABLE_ID_LEN);
pMsg += sizeof(SRemoveMeterMsg); pMsg += sizeof(SRemoveMeterMsg);
msgLen = pMsg - pStart; msgLen = pMsg - pStart;
...@@ -287,7 +287,7 @@ int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) { ...@@ -287,7 +287,7 @@ int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) {
tinet_ntoa(ipstr, pVgroup->vnodeGid[i].ip); tinet_ntoa(ipstr, pVgroup->vnodeGid[i].ip);
mTrace("dnode:%s vid:%d, send remove meter msg, sid:%d status:%d", ipstr, pVgroup->vnodeGid[i].vnode, mTrace("dnode:%s vid:%d, send remove meter msg, sid:%d status:%d", ipstr, pVgroup->vnodeGid[i].vnode,
pMeter->gid.sid, pObj->status); pTable->gid.sid, pObj->status);
} }
pVgroup->lastRemove = timeStamp; pVgroup->lastRemove = timeStamp;
...@@ -295,7 +295,7 @@ int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) { ...@@ -295,7 +295,7 @@ int mgmtSendRemoveMeterMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) {
return 0; return 0;
} }
int mgmtSendAlterStreamMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) { int mgmtSendAlterStreamMsgToDnode(STabObj *pTable, SVgObj *pVgroup) {
SAlterStreamMsg *pAlter; SAlterStreamMsg *pAlter;
char * pMsg, *pStart; char * pMsg, *pStart;
int i, msgLen = 0; int i, msgLen = 0;
...@@ -313,9 +313,9 @@ int mgmtSendAlterStreamMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) { ...@@ -313,9 +313,9 @@ int mgmtSendAlterStreamMsgToDnode(STabObj *pMeter, SVgObj *pVgroup) {
pAlter = (SAlterStreamMsg *)pMsg; pAlter = (SAlterStreamMsg *)pMsg;
pAlter->vnode = htons(pVgroup->vnodeGid[i].vnode); pAlter->vnode = htons(pVgroup->vnodeGid[i].vnode);
pAlter->sid = htonl(pMeter->gid.sid); pAlter->sid = htonl(pTable->gid.sid);
pAlter->uid = pMeter->uid; pAlter->uid = pTable->uid;
pAlter->status = pMeter->status; pAlter->status = pTable->status;
pMsg += sizeof(SAlterStreamMsg); pMsg += sizeof(SAlterStreamMsg);
msgLen = pMsg - pStart; msgLen = pMsg - pStart;
......
此差异已折叠。
...@@ -24,7 +24,7 @@ ...@@ -24,7 +24,7 @@
typedef struct { typedef struct {
uint32_t ip; uint32_t ip;
uint16_t port; uint16_t port;
char user[TSDB_METER_ID_LEN]; char user[TSDB_TABLE_ID_LEN];
} SCDesc; } SCDesc;
typedef struct { typedef struct {
......
...@@ -118,7 +118,7 @@ static void mgmtSetSchemaFromMeters(SSchema *pSchema, STabObj *pMeterObj, uint32 ...@@ -118,7 +118,7 @@ static void mgmtSetSchemaFromMeters(SSchema *pSchema, STabObj *pMeterObj, uint32
static uint32_t mgmtSetMeterTagValue(char *pTags, STabObj *pMetric, STabObj *pMeterObj) { static uint32_t mgmtSetMeterTagValue(char *pTags, STabObj *pMetric, STabObj *pMeterObj) {
SSchema *pTagSchema = (SSchema *)(pMetric->schema + pMetric->numOfColumns * sizeof(SSchema)); SSchema *pTagSchema = (SSchema *)(pMetric->schema + pMetric->numOfColumns * sizeof(SSchema));
char *tagVal = pMeterObj->pTagData + TSDB_METER_ID_LEN; // tag start position char *tagVal = pMeterObj->pTagData + TSDB_TABLE_ID_LEN; // tag start position
uint32_t tagsLen = 0; uint32_t tagsLen = 0;
for (int32_t i = 0; i < pMetric->numOfTags; ++i) { for (int32_t i = 0; i < pMetric->numOfTags; ++i) {
...@@ -224,8 +224,8 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -224,8 +224,8 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
int32_t code = mgmtCreateMeter(pDb, pCreateMsg); int32_t code = mgmtCreateMeter(pDb, pCreateMsg);
char stableName[TSDB_METER_ID_LEN] = {0}; char stableName[TSDB_TABLE_ID_LEN] = {0};
strncpy(stableName, pInfo->tags, TSDB_METER_ID_LEN); strncpy(stableName, pInfo->tags, TSDB_TABLE_ID_LEN);
mTrace("meter:%s is automatically created by %s from %s, code:%d", pCreateMsg->meterId, pConn->pUser->user, mTrace("meter:%s is automatically created by %s from %s, code:%d", pCreateMsg->meterId, pConn->pUser->user,
stableName, code); stableName, code);
...@@ -274,7 +274,7 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -274,7 +274,7 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pMeta->numOfTags = pMeterObj->numOfTags; pMeta->numOfTags = pMeterObj->numOfTags;
pMeta->numOfColumns = htons(pMeterObj->numOfColumns); pMeta->numOfColumns = htons(pMeterObj->numOfColumns);
pMeta->meterType = pMeterObj->meterType; pMeta->tableType = pMeterObj->tableType;
pMsg += sizeof(SMeterMeta); pMsg += sizeof(SMeterMeta);
pSchema = (SSchema *)pMsg; // schema locates at the end of SMeterMeta struct pSchema = (SSchema *)pMsg; // schema locates at the end of SMeterMeta struct
...@@ -361,7 +361,7 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -361,7 +361,7 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
} }
int32_t totalNum = 0; int32_t totalNum = 0;
char tblName[TSDB_METER_ID_LEN]; char tblName[TSDB_TABLE_ID_LEN];
char* nextStr; char* nextStr;
char* pCurMeter = pStart + sizeof(STaosRsp) + sizeof(SMultiMeterInfoMsg) + 1; // 1: ie type byte char* pCurMeter = pStart + sizeof(STaosRsp) + sizeof(SMultiMeterInfoMsg) + 1; // 1: ie type byte
...@@ -414,7 +414,7 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -414,7 +414,7 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pMeta->meta.precision = pDbObj->cfg.precision; pMeta->meta.precision = pDbObj->cfg.precision;
pMeta->meta.numOfTags = pMeterObj->numOfTags; pMeta->meta.numOfTags = pMeterObj->numOfTags;
pMeta->meta.numOfColumns = htons(pMeterObj->numOfColumns); pMeta->meta.numOfColumns = htons(pMeterObj->numOfColumns);
pMeta->meta.meterType = pMeterObj->meterType; pMeta->meta.tableType = pMeterObj->tableType;
pCurMeter += sizeof(SMultiMeterMeta); pCurMeter += sizeof(SMultiMeterMeta);
pSchema = (SSchema *)pCurMeter; // schema locates at the end of SMeterMeta struct pSchema = (SSchema *)pCurMeter; // schema locates at the end of SMeterMeta struct
...@@ -494,21 +494,21 @@ _exit_code: ...@@ -494,21 +494,21 @@ _exit_code:
} }
int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
SMetricMetaMsg *pMetricMetaMsg = (SMetricMetaMsg *)pMsg; SSuperTableMetaMsg *pSuperTableMetaMsg = (SSuperTableMetaMsg *)pMsg;
STabObj * pMetric; STabObj * pMetric;
STaosRsp * pRsp; STaosRsp * pRsp;
char * pStart; char * pStart;
pMetricMetaMsg->numOfMeters = htonl(pMetricMetaMsg->numOfMeters); pSuperTableMetaMsg->numOfMeters = htonl(pSuperTableMetaMsg->numOfMeters);
pMetricMetaMsg->join = htonl(pMetricMetaMsg->join); pSuperTableMetaMsg->join = htonl(pSuperTableMetaMsg->join);
pMetricMetaMsg->joinCondLen = htonl(pMetricMetaMsg->joinCondLen); pSuperTableMetaMsg->joinCondLen = htonl(pSuperTableMetaMsg->joinCondLen);
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) { for (int32_t i = 0; i < pSuperTableMetaMsg->numOfMeters; ++i) {
pMetricMetaMsg->metaElem[i] = htonl(pMetricMetaMsg->metaElem[i]); pSuperTableMetaMsg->metaElem[i] = htonl(pSuperTableMetaMsg->metaElem[i]);
} }
SMetricMetaElemMsg *pElem = (SMetricMetaElemMsg *)(((char *)pMetricMetaMsg) + pMetricMetaMsg->metaElem[0]); SMetricMetaElemMsg *pElem = (SMetricMetaElemMsg *)(((char *)pSuperTableMetaMsg) + pSuperTableMetaMsg->metaElem[0]);
pMetric = mgmtGetTable(pElem->meterId); pMetric = mgmtGetTable(pElem->meterId);
SDbObj *pDb = NULL; SDbObj *pDb = NULL;
...@@ -531,7 +531,7 @@ int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -531,7 +531,7 @@ int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
msgLen = pMsg - pStart; msgLen = pMsg - pStart;
} else { } else {
msgLen = mgmtRetrieveMetricMeta(pConn, &pStart, pMetricMetaMsg); msgLen = mgmtRetrieveMetricMeta(pConn, &pStart, pSuperTableMetaMsg);
if (msgLen <= 0) { if (msgLen <= 0) {
taosSendSimpleRsp(pConn->thandle, TSDB_MSG_TYPE_METRIC_META_RSP, TSDB_CODE_SERV_OUT_OF_MEMORY); taosSendSimpleRsp(pConn->thandle, TSDB_MSG_TYPE_METRIC_META_RSP, TSDB_CODE_SERV_OUT_OF_MEMORY);
return 0; return 0;
...@@ -1076,11 +1076,11 @@ int mgmtProcessCreateTableMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -1076,11 +1076,11 @@ int mgmtProcessCreateTableMsg(char *pMsg, int msgLen, SConnObj *pConn) {
} else if (code != TSDB_CODE_SUCCESS) { } else if (code != TSDB_CODE_SUCCESS) {
if (code == TSDB_CODE_TABLE_ALREADY_EXIST) { // table already created when the second attempt to create table if (code == TSDB_CODE_TABLE_ALREADY_EXIST) { // table already created when the second attempt to create table
STabObj* pMeter = mgmtGetTable(pCreate->meterId); STabObj* pTable = mgmtGetTable(pCreate->meterId);
assert(pMeter != NULL); assert(pTable != NULL);
mWarn("table:%s, table already created, failed to create table, ts:%" PRId64 ", code:%d", pCreate->meterId, mWarn("table:%s, table already created, failed to create table, ts:%" PRId64 ", code:%d", pCreate->meterId,
pMeter->createdTime, code); pTable->createdTime, code);
} else { // other errors } else { // other errors
mError("table:%s, failed to create table, code:%d", pCreate->meterId, code); mError("table:%s, failed to create table, code:%d", pCreate->meterId, code);
} }
......
...@@ -23,6 +23,12 @@ ...@@ -23,6 +23,12 @@
#include "tsqlfunction.h" #include "tsqlfunction.h"
#include "vnodeTagMgmt.h" #include "vnodeTagMgmt.h"
typedef struct SSyntaxTreeFilterSupporter {
SSchema* pTagSchema;
int32_t numOfTags;
int32_t optr;
} SSyntaxTreeFilterSupporter;
typedef struct SJoinSupporter { typedef struct SJoinSupporter {
void** val; void** val;
void** pTabObjs; void** pTabObjs;
...@@ -41,6 +47,7 @@ typedef struct SMeterNameFilterSupporter { ...@@ -41,6 +47,7 @@ typedef struct SMeterNameFilterSupporter {
} SMeterNameFilterSupporter; } SMeterNameFilterSupporter;
static void tansformQueryResult(tQueryResultset* pRes); static void tansformQueryResult(tQueryResultset* pRes);
static bool tSkipListNodeFilterCallback(const void *pNode, void *param);
static int32_t tabObjVGIDComparator(const void* pLeft, const void* pRight) { static int32_t tabObjVGIDComparator(const void* pLeft, const void* pRight) {
STabObj* p1 = *(STabObj**)pLeft; STabObj* p1 = *(STabObj**)pLeft;
...@@ -82,7 +89,7 @@ static int32_t tabObjResultComparator(const void* p1, const void* p2, void* para ...@@ -82,7 +89,7 @@ static int32_t tabObjResultComparator(const void* p1, const void* p2, void* para
f1 = pNode1->meterId; f1 = pNode1->meterId;
f2 = pNode2->meterId; f2 = pNode2->meterId;
schema.type = TSDB_DATA_TYPE_BINARY; schema.type = TSDB_DATA_TYPE_BINARY;
schema.bytes = TSDB_METER_ID_LEN; schema.bytes = TSDB_TABLE_ID_LEN;
} else { } else {
f1 = mgmtTableGetTag(pNode1, colIdx, NULL); f1 = mgmtTableGetTag(pNode1, colIdx, NULL);
f2 = mgmtTableGetTag(pNode2, colIdx, &schema); f2 = mgmtTableGetTag(pNode2, colIdx, &schema);
...@@ -104,15 +111,15 @@ static int32_t tabObjResultComparator(const void* p1, const void* p2, void* para ...@@ -104,15 +111,15 @@ static int32_t tabObjResultComparator(const void* p1, const void* p2, void* para
* update the tag order index according to the tags column index. The tags column index needs to be checked one-by-one, * update the tag order index according to the tags column index. The tags column index needs to be checked one-by-one,
* since the normal columns may be passed to server for handling the group by on status column. * since the normal columns may be passed to server for handling the group by on status column.
* *
* @param pMetricMetaMsg * @param pSuperTableMetaMsg
* @param tableIndex * @param tableIndex
* @param pOrderIndexInfo * @param pOrderIndexInfo
* @param numOfTags * @param numOfTags
*/ */
static void mgmtUpdateOrderTagColIndex(SMetricMetaMsg* pMetricMetaMsg, int32_t tableIndex, tOrderIdx* pOrderIndexInfo, static void mgmtUpdateOrderTagColIndex(SSuperTableMetaMsg* pSuperTableMetaMsg, int32_t tableIndex, tOrderIdx* pOrderIndexInfo,
int32_t numOfTags) { int32_t numOfTags) {
SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pMetricMetaMsg + pMetricMetaMsg->metaElem[tableIndex]); SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pSuperTableMetaMsg + pSuperTableMetaMsg->metaElem[tableIndex]);
SColIndexEx* groupColumnList = (SColIndexEx*)((char*)pMetricMetaMsg + pElem->groupbyTagColumnList); SColIndexEx* groupColumnList = (SColIndexEx*)((char*)pSuperTableMetaMsg + pElem->groupbyTagColumnList);
int32_t numOfGroupbyTags = 0; int32_t numOfGroupbyTags = 0;
for (int32_t i = 0; i < pElem->numOfGroupCols; ++i) { for (int32_t i = 0; i < pElem->numOfGroupCols; ++i) {
...@@ -127,12 +134,12 @@ static void mgmtUpdateOrderTagColIndex(SMetricMetaMsg* pMetricMetaMsg, int32_t t ...@@ -127,12 +134,12 @@ static void mgmtUpdateOrderTagColIndex(SMetricMetaMsg* pMetricMetaMsg, int32_t t
} }
// todo merge sort function with losertree used // todo merge sort function with losertree used
void mgmtReorganizeMetersInMetricMeta(SMetricMetaMsg* pMetricMetaMsg, int32_t tableIndex, tQueryResultset* pRes) { void mgmtReorganizeMetersInMetricMeta(SSuperTableMetaMsg* pSuperTableMetaMsg, int32_t tableIndex, tQueryResultset* pRes) {
if (pRes->num <= 0) { // no result, no need to pagination if (pRes->num <= 0) { // no result, no need to pagination
return; return;
} }
SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pMetricMetaMsg + pMetricMetaMsg->metaElem[tableIndex]); SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pSuperTableMetaMsg + pSuperTableMetaMsg->metaElem[tableIndex]);
STabObj* pMetric = mgmtGetTable(pElem->meterId); STabObj* pMetric = mgmtGetTable(pElem->meterId);
SSchema* pTagSchema = (SSchema*)(pMetric->schema + pMetric->numOfColumns * sizeof(SSchema)); SSchema* pTagSchema = (SSchema*)(pMetric->schema + pMetric->numOfColumns * sizeof(SSchema));
...@@ -149,7 +156,7 @@ void mgmtReorganizeMetersInMetricMeta(SMetricMetaMsg* pMetricMetaMsg, int32_t ta ...@@ -149,7 +156,7 @@ void mgmtReorganizeMetersInMetricMeta(SMetricMetaMsg* pMetricMetaMsg, int32_t ta
int32_t* startPos = NULL; int32_t* startPos = NULL;
int32_t numOfSubset = 1; int32_t numOfSubset = 1;
mgmtUpdateOrderTagColIndex(pMetricMetaMsg, tableIndex, &descriptor->orderIdx, pMetric->numOfTags); mgmtUpdateOrderTagColIndex(pSuperTableMetaMsg, tableIndex, &descriptor->orderIdx, pMetric->numOfTags);
if (descriptor->orderIdx.numOfOrderedCols > 0) { if (descriptor->orderIdx.numOfOrderedCols > 0) {
tQSortEx(pRes->pRes, POINTER_BYTES, 0, pRes->num - 1, descriptor, tabObjResultComparator); tQSortEx(pRes->pRes, POINTER_BYTES, 0, pRes->num - 1, descriptor, tabObjResultComparator);
startPos = calculateSubGroup(pRes->pRes, pRes->num, &numOfSubset, descriptor, tabObjResultComparator); startPos = calculateSubGroup(pRes->pRes, pRes->num, &numOfSubset, descriptor, tabObjResultComparator);
...@@ -193,7 +200,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p ...@@ -193,7 +200,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p
} }
/* not a table created from metric, ignore */ /* not a table created from metric, ignore */
if (pMeterObj->meterType != TSDB_TABLE_TYPE_CREATE_FROM_STABLE) { if (pMeterObj->tableType != TSDB_TABLE_TYPE_CREATE_FROM_STABLE) {
continue; continue;
} }
...@@ -202,7 +209,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p ...@@ -202,7 +209,7 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p
* uid, so compare according to meterid * uid, so compare according to meterid
*/ */
STabObj* parentMetric = mgmtGetTable(pMeterObj->pTagData); STabObj* parentMetric = mgmtGetTable(pMeterObj->pTagData);
if (strncasecmp(parentMetric->meterId, pMetric->meterId, TSDB_METER_ID_LEN) != 0 || if (strncasecmp(parentMetric->meterId, pMetric->meterId, TSDB_TABLE_ID_LEN) != 0 ||
(parentMetric->uid != pMetric->uid)) { (parentMetric->uid != pMetric->uid)) {
continue; continue;
} }
...@@ -214,13 +221,13 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p ...@@ -214,13 +221,13 @@ static void mgmtRetrieveByMeterName(tQueryResultset* pRes, char* str, STabObj* p
static bool mgmtTablenameFilterCallback(tSkipListNode* pNode, void* param) { static bool mgmtTablenameFilterCallback(tSkipListNode* pNode, void* param) {
SMeterNameFilterSupporter* pSupporter = (SMeterNameFilterSupporter*)param; SMeterNameFilterSupporter* pSupporter = (SMeterNameFilterSupporter*)param;
char name[TSDB_METER_ID_LEN] = {0}; char name[TSDB_TABLE_ID_LEN] = {0};
// pattern compare for meter name // pattern compare for meter name
STabObj* pMeterObj = (STabObj*)pNode->pData; STabObj* pMeterObj = (STabObj*)pNode->pData;
extractTableName(pMeterObj->meterId, name); extractTableName(pMeterObj->meterId, name);
return patternMatch(pSupporter->pattern, name, TSDB_METER_ID_LEN, &pSupporter->info) == TSDB_PATTERN_MATCH; return patternMatch(pSupporter->pattern, name, TSDB_TABLE_ID_LEN, &pSupporter->info) == TSDB_PATTERN_MATCH;
} }
static void mgmtRetrieveFromLikeOptr(tQueryResultset* pRes, const char* str, STabObj* pMetric) { static void mgmtRetrieveFromLikeOptr(tQueryResultset* pRes, const char* str, STabObj* pMetric) {
...@@ -282,8 +289,8 @@ UNUSED_FUNC static bool mgmtJoinFilterCallback(tSkipListNode* pNode, void* param ...@@ -282,8 +289,8 @@ UNUSED_FUNC static bool mgmtJoinFilterCallback(tSkipListNode* pNode, void* param
return false; return false;
} }
static void orderResult(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes, int16_t colIndex, int32_t tableIndex) { static void orderResult(SSuperTableMetaMsg* pSuperTableMetaMsg, tQueryResultset* pRes, int16_t colIndex, int32_t tableIndex) {
SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pMetricMetaMsg + pMetricMetaMsg->metaElem[tableIndex]); SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pSuperTableMetaMsg + pSuperTableMetaMsg->metaElem[tableIndex]);
tOrderDescriptor* descriptor = tOrderDescriptor* descriptor =
(tOrderDescriptor*)calloc(1, sizeof(tOrderDescriptor) + sizeof(int32_t) * 1); // only one column for join (tOrderDescriptor*)calloc(1, sizeof(tOrderDescriptor) + sizeof(int32_t) * 1); // only one column for join
...@@ -322,13 +329,13 @@ static int32_t mgmtCheckForDuplicateTagValue(tQueryResultset* pRes, int32_t inde ...@@ -322,13 +329,13 @@ static int32_t mgmtCheckForDuplicateTagValue(tQueryResultset* pRes, int32_t inde
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) { int32_t mgmtDoJoin(SSuperTableMetaMsg* pSuperTableMetaMsg, tQueryResultset* pRes) {
if (pMetricMetaMsg->numOfMeters == 1) { if (pSuperTableMetaMsg->numOfMeters == 1) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
bool allEmpty = false; bool allEmpty = false;
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) { for (int32_t i = 0; i < pSuperTableMetaMsg->numOfMeters; ++i) {
if (pRes[i].num == 0) { // all results are empty if one of them is empty if (pRes[i].num == 0) { // all results are empty if one of them is empty
allEmpty = true; allEmpty = true;
break; break;
...@@ -336,7 +343,7 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) { ...@@ -336,7 +343,7 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
} }
if (allEmpty) { if (allEmpty) {
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) { for (int32_t i = 0; i < pSuperTableMetaMsg->numOfMeters; ++i) {
pRes[i].num = 0; pRes[i].num = 0;
tfree(pRes[i].pRes); tfree(pRes[i].pRes);
} }
...@@ -344,15 +351,15 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) { ...@@ -344,15 +351,15 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
char* cond = (char*)pMetricMetaMsg + pMetricMetaMsg->join; char* cond = (char*)pSuperTableMetaMsg + pSuperTableMetaMsg->join;
char left[TSDB_METER_ID_LEN + 1] = {0}; char left[TSDB_TABLE_ID_LEN + 1] = {0};
strcpy(left, cond); strcpy(left, cond);
int16_t leftTagColIndex = *(int16_t*)(cond + TSDB_METER_ID_LEN); int16_t leftTagColIndex = *(int16_t*)(cond + TSDB_TABLE_ID_LEN);
char right[TSDB_METER_ID_LEN + 1] = {0}; char right[TSDB_TABLE_ID_LEN + 1] = {0};
strcpy(right, cond + TSDB_METER_ID_LEN + sizeof(int16_t)); strcpy(right, cond + TSDB_TABLE_ID_LEN + sizeof(int16_t));
int16_t rightTagColIndex = *(int16_t*)(cond + TSDB_METER_ID_LEN * 2 + sizeof(int16_t)); int16_t rightTagColIndex = *(int16_t*)(cond + TSDB_TABLE_ID_LEN * 2 + sizeof(int16_t));
STabObj* pLeftMetric = mgmtGetTable(left); STabObj* pLeftMetric = mgmtGetTable(left);
STabObj* pRightMetric = mgmtGetTable(right); STabObj* pRightMetric = mgmtGetTable(right);
...@@ -361,7 +368,7 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) { ...@@ -361,7 +368,7 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
int32_t leftIndex = 0; int32_t leftIndex = 0;
int32_t rightIndex = 0; int32_t rightIndex = 0;
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) { for (int32_t i = 0; i < pSuperTableMetaMsg->numOfMeters; ++i) {
STabObj* pObj = (STabObj*)pRes[i].pRes[0]; STabObj* pObj = (STabObj*)pRes[i].pRes[0];
STabObj* pMetric1 = mgmtGetTable(pObj->pTagData); STabObj* pMetric1 = mgmtGetTable(pObj->pTagData);
if (pMetric1 == pLeftMetric) { if (pMetric1 == pLeftMetric) {
...@@ -371,8 +378,8 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) { ...@@ -371,8 +378,8 @@ int32_t mgmtDoJoin(SMetricMetaMsg* pMetricMetaMsg, tQueryResultset* pRes) {
} }
} }
orderResult(pMetricMetaMsg, &pRes[leftIndex], leftTagColIndex, leftIndex); orderResult(pSuperTableMetaMsg, &pRes[leftIndex], leftTagColIndex, leftIndex);
orderResult(pMetricMetaMsg, &pRes[rightIndex], rightTagColIndex, rightIndex); orderResult(pSuperTableMetaMsg, &pRes[rightIndex], rightTagColIndex, rightIndex);
int32_t i = 0; int32_t i = 0;
int32_t j = 0; int32_t j = 0;
...@@ -727,7 +734,7 @@ static int32_t mgmtFilterMeterByIndex(STabObj* pMetric, tQueryResultset* pRes, c ...@@ -727,7 +734,7 @@ static int32_t mgmtFilterMeterByIndex(STabObj* pMetric, tQueryResultset* pRes, c
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int mgmtRetrieveMetersFromMetric(SMetricMetaMsg* pMsg, int32_t tableIndex, tQueryResultset* pRes) { int32_t mgmtRetrieveMetersFromSuperTable(SSuperTableMetaMsg* pMsg, int32_t tableIndex, tQueryResultset* pRes) {
SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pMsg + pMsg->metaElem[tableIndex]); SMetricMetaElemMsg* pElem = (SMetricMetaElemMsg*)((char*)pMsg + pMsg->metaElem[tableIndex]);
STabObj* pMetric = mgmtGetTable(pElem->meterId); STabObj* pMetric = mgmtGetTable(pElem->meterId);
char* pCond = NULL; char* pCond = NULL;
...@@ -806,11 +813,11 @@ int mgmtRetrieveMetersFromMetric(SMetricMetaMsg* pMsg, int32_t tableIndex, tQuer ...@@ -806,11 +813,11 @@ int mgmtRetrieveMetersFromMetric(SMetricMetaMsg* pMsg, int32_t tableIndex, tQuer
} }
// todo refactor!!!!! // todo refactor!!!!!
static char* getTagValueFromMeter(STabObj* pMeter, int32_t offset, int32_t len, char* param) { static char* getTagValueFromMeter(STabObj* pTable, int32_t offset, int32_t len, char* param) {
if (offset == TSDB_TBNAME_COLUMN_INDEX) { if (offset == TSDB_TBNAME_COLUMN_INDEX) {
extractTableName(pMeter->meterId, param); extractTableName(pTable->meterId, param);
} else { } else {
char* tags = pMeter->pTagData + offset + TSDB_METER_ID_LEN; // tag start position char* tags = pTable->pTagData + offset + TSDB_TABLE_ID_LEN; // tag start position
memcpy(param, tags, len); // make sure the value is null-terminated string memcpy(param, tags, len); // make sure the value is null-terminated string
} }
...@@ -820,11 +827,11 @@ static char* getTagValueFromMeter(STabObj* pMeter, int32_t offset, int32_t len, ...@@ -820,11 +827,11 @@ static char* getTagValueFromMeter(STabObj* pMeter, int32_t offset, int32_t len,
bool tSkipListNodeFilterCallback(const void* pNode, void* param) { bool tSkipListNodeFilterCallback(const void* pNode, void* param) {
tQueryInfo* pInfo = (tQueryInfo*)param; tQueryInfo* pInfo = (tQueryInfo*)param;
STabObj* pMeter = (STabObj*)(((tSkipListNode*)pNode)->pData); STabObj* pTable = (STabObj*)(((tSkipListNode*)pNode)->pData);
char buf[TSDB_MAX_TAGS_LEN] = {0}; char buf[TSDB_MAX_TAGS_LEN] = {0};
char* val = getTagValueFromMeter(pMeter, pInfo->offset, pInfo->sch.bytes, buf); char* val = getTagValueFromMeter(pTable, pInfo->offset, pInfo->sch.bytes, buf);
int8_t type = pInfo->sch.type; int8_t type = pInfo->sch.type;
int32_t ret = 0; int32_t ret = 0;
......
...@@ -15,19 +15,16 @@ ...@@ -15,19 +15,16 @@
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
#include "mnode.h" #include "mnode.h"
#include "mgmtUtil.h" #include "mgmtUtil.h"
#include "textbuffer.h"
#include "tschemautil.h" #include "tschemautil.h"
#include "tsqlfunction.h"
bool mgmtTableCreateFromSuperTable(STabObj* pTableObj) { bool mgmtTableCreateFromSuperTable(STabObj* pTableObj) {
return pTableObj->meterType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE; return pTableObj->tableType == TSDB_TABLE_TYPE_CREATE_FROM_STABLE;
} }
bool mgmtIsSuperTable(STabObj* pTableObj) { bool mgmtIsSuperTable(STabObj* pTableObj) {
return pTableObj->meterType == TSDB_TABLE_TYPE_SUPER_TABLE; return pTableObj->tableType == TSDB_TABLE_TYPE_SUPER_TABLE;
} }
bool mgmtIsNormalTable(STabObj* pTableObj) { bool mgmtIsNormalTable(STabObj* pTableObj) {
...@@ -37,7 +34,7 @@ bool mgmtIsNormalTable(STabObj* pTableObj) { ...@@ -37,7 +34,7 @@ bool mgmtIsNormalTable(STabObj* pTableObj) {
/** /**
* TODO: the tag offset value should be kept in memory to avoid dynamically calculating the value * TODO: the tag offset value should be kept in memory to avoid dynamically calculating the value
* *
* @param pMeter * @param pTable
* @param col * @param col
* @param pTagColSchema * @param pTagColSchema
* @return * @return
...@@ -48,7 +45,7 @@ char* mgmtTableGetTag(STabObj* pTable, int32_t col, SSchema* pTagColSchema) { ...@@ -48,7 +45,7 @@ char* mgmtTableGetTag(STabObj* pTable, int32_t col, SSchema* pTagColSchema) {
} }
STabObj* pSuperTable = mgmtGetTable(pTable->pTagData); STabObj* pSuperTable = mgmtGetTable(pTable->pTagData);
int32_t offset = mgmtGetTagsLength(pSuperTable, col) + TSDB_METER_ID_LEN; int32_t offset = mgmtGetTagsLength(pSuperTable, col) + TSDB_TABLE_ID_LEN;
assert(offset > 0); assert(offset > 0);
if (pTagColSchema != NULL) { if (pTagColSchema != NULL) {
......
...@@ -168,13 +168,13 @@ SVgObj *mgmtCreateVgroup(SDbObj *pDb) { ...@@ -168,13 +168,13 @@ SVgObj *mgmtCreateVgroup(SDbObj *pDb) {
} }
int mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup) { int mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup) {
STabObj *pMeter; STabObj *pTable;
if (pVgroup->numOfMeters > 0) { if (pVgroup->numOfMeters > 0) {
for (int i = 0; i < pDb->cfg.maxSessions; ++i) { for (int i = 0; i < pDb->cfg.maxSessions; ++i) {
if (pVgroup->meterList != NULL) { if (pVgroup->meterList != NULL) {
pMeter = pVgroup->meterList[i]; pTable = pVgroup->meterList[i];
if (pMeter) mgmtDropMeter(pDb, pMeter->meterId, 0); if (pTable) mgmtDropMeter(pDb, pTable->meterId, 0);
} }
} }
} }
...@@ -239,14 +239,14 @@ int mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn) { ...@@ -239,14 +239,14 @@ int mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn) {
int maxReplica = 0; int maxReplica = 0;
SVgObj *pVgroup = NULL; SVgObj *pVgroup = NULL;
STabObj *pMeter = NULL; STabObj *pTable = NULL;
if (pShow->payloadLen > 0 ) { if (pShow->payloadLen > 0 ) {
pMeter = mgmtGetTable(pShow->payload); pTable = mgmtGetTable(pShow->payload);
if (NULL == pMeter) { if (NULL == pTable) {
return TSDB_CODE_INVALID_METER_ID; return TSDB_CODE_INVALID_METER_ID;
} }
pVgroup = mgmtGetVgroup(pMeter->gid.vgId); pVgroup = mgmtGetVgroup(pTable->gid.vgId);
if (NULL == pVgroup) return TSDB_CODE_INVALID_METER_ID; if (NULL == pVgroup) return TSDB_CODE_INVALID_METER_ID;
maxReplica = pVgroup->numOfVnodes > maxReplica ? pVgroup->numOfVnodes : maxReplica; maxReplica = pVgroup->numOfVnodes > maxReplica ? pVgroup->numOfVnodes : maxReplica;
...@@ -292,7 +292,7 @@ int mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn) { ...@@ -292,7 +292,7 @@ int mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, SConnObj *pConn) {
pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1]; pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
if (NULL == pMeter) { if (NULL == pTable) {
pShow->numOfRows = pDb->numOfVgroups; pShow->numOfRows = pDb->numOfVgroups;
pShow->pNode = pDb->pHead; pShow->pNode = pDb->pHead;
} else { } else {
......
...@@ -188,7 +188,7 @@ void dnodeBuildMonitorSql(char *sql, int cmd) { ...@@ -188,7 +188,7 @@ void dnodeBuildMonitorSql(char *sql, int cmd) {
snprintf(sql, SQL_LENGTH, snprintf(sql, SQL_LENGTH,
"create table if not exists %s.slowquery(ts timestamp, username " "create table if not exists %s.slowquery(ts timestamp, username "
"binary(%d), created_time timestamp, time bigint, sql binary(%d))", "binary(%d), created_time timestamp, time bigint, sql binary(%d))",
tsMonitorDbName, TSDB_METER_ID_LEN, TSDB_SHOW_SQL_LEN); tsMonitorDbName, TSDB_TABLE_ID_LEN, TSDB_SHOW_SQL_LEN);
} else if (cmd == MONITOR_CMD_CREATE_TB_LOG) { } else if (cmd == MONITOR_CMD_CREATE_TB_LOG) {
snprintf(sql, SQL_LENGTH, snprintf(sql, SQL_LENGTH,
"create table if not exists %s.log(ts timestamp, level tinyint, " "create table if not exists %s.log(ts timestamp, level tinyint, "
......
...@@ -722,7 +722,7 @@ int taosSetSecurityInfo(int chann, int sid, char *id, int spi, int encrypt, char ...@@ -722,7 +722,7 @@ int taosSetSecurityInfo(int chann, int sid, char *id, int spi, int encrypt, char
pConn->encrypt = encrypt; pConn->encrypt = encrypt;
memcpy(pConn->secret, pConn->secret, TSDB_KEY_LEN); memcpy(pConn->secret, pConn->secret, TSDB_KEY_LEN);
memcpy(pConn->cipheringKey, ckey, TSDB_KEY_LEN); memcpy(pConn->cipheringKey, ckey, TSDB_KEY_LEN);
memcpy(pConn->meterId, id, TSDB_METER_ID_LEN); memcpy(pConn->meterId, id, TSDB_TABLE_ID_LEN);
*/ */
return -1; return -1;
} }
......
...@@ -19,7 +19,7 @@ ...@@ -19,7 +19,7 @@
#define MAX_STR_LEN 40 #define MAX_STR_LEN 40
typedef struct _str_node_t { typedef struct _str_node_t {
char string[TSDB_METER_ID_LEN]; char string[TSDB_TABLE_ID_LEN];
int hash; int hash;
struct _str_node_t *prev; struct _str_node_t *prev;
struct _str_node_t *next; struct _str_node_t *next;
......
...@@ -158,7 +158,7 @@ typedef struct SColumn { ...@@ -158,7 +158,7 @@ typedef struct SColumn {
typedef struct _meter_obj { typedef struct _meter_obj {
uint64_t uid; uint64_t uid;
char meterId[TSDB_METER_ID_LEN]; char meterId[TSDB_TABLE_ID_LEN];
int sid; int sid;
short vnode; short vnode;
short numOfColumns; short numOfColumns;
......
...@@ -253,7 +253,7 @@ typedef struct SMeterQuerySupportObj { ...@@ -253,7 +253,7 @@ typedef struct SMeterQuerySupportObj {
typedef struct _qinfo { typedef struct _qinfo {
uint64_t signature; uint64_t signature;
int32_t refCount; // QInfo reference count, when the value is 0, it can be released safely int32_t refCount; // QInfo reference count, when the value is 0, it can be released safely
char user[TSDB_METER_ID_LEN + 1]; char user[TSDB_TABLE_ID_LEN + 1];
char sql[TSDB_SHOW_SQL_LEN]; char sql[TSDB_SHOW_SQL_LEN];
uint8_t stream; uint8_t stream;
uint16_t port; uint16_t port;
......
...@@ -492,7 +492,7 @@ void *vnodeCommitMultiToFile(SVnodeObj *pVnode, int ssid, int esid) { ...@@ -492,7 +492,7 @@ void *vnodeCommitMultiToFile(SVnodeObj *pVnode, int ssid, int esid) {
SMeterObj * pObj = NULL; SMeterObj * pObj = NULL;
SCompInfo compInfo = {0}; SCompInfo compInfo = {0};
SCompHeader * pHeader; SCompHeader * pHeader;
SMeterInfo * meterInfo = NULL, *pMeter = NULL; SMeterInfo * meterInfo = NULL, *pTable = NULL;
SQuery query; SQuery query;
SColumnInfoEx colList[TSDB_MAX_COLUMNS] = {0}; SColumnInfoEx colList[TSDB_MAX_COLUMNS] = {0};
SSqlFunctionExpr pExprs[TSDB_MAX_COLUMNS] = {0}; SSqlFunctionExpr pExprs[TSDB_MAX_COLUMNS] = {0};
...@@ -617,7 +617,7 @@ _again: ...@@ -617,7 +617,7 @@ _again:
continue; continue;
} }
pMeter = meterInfo + sid; pTable = meterInfo + sid;
pHeader = ((SCompHeader *)tmem) + sid; pHeader = ((SCompHeader *)tmem) + sid;
if (pVnode->hfd > 0) { if (pVnode->hfd > 0) {
...@@ -633,18 +633,18 @@ _again: ...@@ -633,18 +633,18 @@ _again:
goto _over; goto _over;
} else { } else {
if (pObj->uid == compInfo.uid) { if (pObj->uid == compInfo.uid) {
pMeter->oldNumOfBlocks = compInfo.numOfBlocks; pTable->oldNumOfBlocks = compInfo.numOfBlocks;
pMeter->oldCompBlockOffset = pHeader->compInfoOffset + sizeof(SCompInfo); pTable->oldCompBlockOffset = pHeader->compInfoOffset + sizeof(SCompInfo);
pMeter->last = compInfo.last; pTable->last = compInfo.last;
if (compInfo.numOfBlocks > maxOldBlocks) maxOldBlocks = compInfo.numOfBlocks; if (compInfo.numOfBlocks > maxOldBlocks) maxOldBlocks = compInfo.numOfBlocks;
if (pMeter->last) { if (pTable->last) {
lseek(pVnode->hfd, sizeof(SCompBlock) * (compInfo.numOfBlocks - 1), SEEK_CUR); lseek(pVnode->hfd, sizeof(SCompBlock) * (compInfo.numOfBlocks - 1), SEEK_CUR);
read(pVnode->hfd, &pMeter->lastBlock, sizeof(SCompBlock)); read(pVnode->hfd, &pTable->lastBlock, sizeof(SCompBlock));
} }
} else { } else {
dTrace("vid:%d sid:%d id:%s, uid:%" PRIu64 " is not matched with old:%" PRIu64 ", old data will be thrown away", dTrace("vid:%d sid:%d id:%s, uid:%" PRIu64 " is not matched with old:%" PRIu64 ", old data will be thrown away",
vnode, sid, pObj->meterId, pObj->uid, compInfo.uid); vnode, sid, pObj->meterId, pObj->uid, compInfo.uid);
pMeter->oldNumOfBlocks = 0; pTable->oldNumOfBlocks = 0;
} }
} }
} else { } else {
...@@ -669,8 +669,8 @@ _again: ...@@ -669,8 +669,8 @@ _again:
pObj->pointsPerFileBlock * pObj->schema[col - 1].bytes + EXTRA_BYTES + sizeof(TSCKSUM)); pObj->pointsPerFileBlock * pObj->schema[col - 1].bytes + EXTRA_BYTES + sizeof(TSCKSUM));
} }
pMeter = meterInfo + sid; pTable = meterInfo + sid;
pMeter->tempHeadOffset = headLen; pTable->tempHeadOffset = headLen;
memset(&query, 0, sizeof(query)); memset(&query, 0, sizeof(query));
query.colList = colList; query.colList = colList;
...@@ -690,27 +690,27 @@ _again: ...@@ -690,27 +690,27 @@ _again:
pointsReadLast = 0; pointsReadLast = 0;
// last block is at last file // last block is at last file
if (pMeter->last) { if (pTable->last) {
if ((pMeter->lastBlock.sversion != pObj->sversion) || (query.over)) { if ((pTable->lastBlock.sversion != pObj->sversion) || (query.over)) {
// TODO : Check the correctness of this code. write the last block to // TODO : Check the correctness of this code. write the last block to
// .data file // .data file
pCompBlock = (SCompBlock *)(hmem + headLen); pCompBlock = (SCompBlock *)(hmem + headLen);
assert(dmem - (char *)pCompBlock >= sizeof(SCompBlock)); assert(dmem - (char *)pCompBlock >= sizeof(SCompBlock));
*pCompBlock = pMeter->lastBlock; *pCompBlock = pTable->lastBlock;
if (pMeter->lastBlock.sversion != pObj->sversion) { if (pTable->lastBlock.sversion != pObj->sversion) {
pCompBlock->last = 0; pCompBlock->last = 0;
pCompBlock->offset = lseek(pVnode->dfd, 0, SEEK_END); pCompBlock->offset = lseek(pVnode->dfd, 0, SEEK_END);
pMeter->last = 0; pTable->last = 0;
lseek(pVnode->lfd, pMeter->lastBlock.offset, SEEK_SET); lseek(pVnode->lfd, pTable->lastBlock.offset, SEEK_SET);
tsendfile(pVnode->dfd, pVnode->lfd, NULL, pMeter->lastBlock.len); tsendfile(pVnode->dfd, pVnode->lfd, NULL, pTable->lastBlock.len);
pVnode->dfSize = pCompBlock->offset + pMeter->lastBlock.len; pVnode->dfSize = pCompBlock->offset + pTable->lastBlock.len;
} else { } else {
if (ssid == 0) { if (ssid == 0) {
assert(pCompBlock->last && pVnode->tfd != -1); assert(pCompBlock->last && pVnode->tfd != -1);
pCompBlock->offset = lseek(pVnode->tfd, 0, SEEK_END); pCompBlock->offset = lseek(pVnode->tfd, 0, SEEK_END);
lseek(pVnode->lfd, pMeter->lastBlock.offset, SEEK_SET); lseek(pVnode->lfd, pTable->lastBlock.offset, SEEK_SET);
tsendfile(pVnode->tfd, pVnode->lfd, NULL, pMeter->lastBlock.len); tsendfile(pVnode->tfd, pVnode->lfd, NULL, pTable->lastBlock.len);
pVnode->lfSize = pCompBlock->offset + pMeter->lastBlock.len; pVnode->lfSize = pCompBlock->offset + pTable->lastBlock.len;
} else { } else {
assert(pVnode->tfd == -1); assert(pVnode->tfd == -1);
} }
...@@ -718,12 +718,12 @@ _again: ...@@ -718,12 +718,12 @@ _again:
} }
headLen += sizeof(SCompBlock); headLen += sizeof(SCompBlock);
pMeter->newNumOfBlocks++; pTable->newNumOfBlocks++;
} else { } else {
// read last block into memory // read last block into memory
if (vnodeReadLastBlockToMem(pObj, &pMeter->lastBlock, data) < 0) goto _over; if (vnodeReadLastBlockToMem(pObj, &pTable->lastBlock, data) < 0) goto _over;
pMeter->last = 0; pTable->last = 0;
pointsReadLast = pMeter->lastBlock.numOfPoints; pointsReadLast = pTable->lastBlock.numOfPoints;
query.over = 0; query.over = 0;
headInfo.totalStorage -= (pointsReadLast * pObj->bytesPerPoint); headInfo.totalStorage -= (pointsReadLast * pObj->bytesPerPoint);
...@@ -731,8 +731,8 @@ _again: ...@@ -731,8 +731,8 @@ _again:
pObj->vnode, pObj->sid, pObj->meterId, pointsReadLast); pObj->vnode, pObj->sid, pObj->meterId, pointsReadLast);
} }
pMeter->changed = 1; pTable->changed = 1;
pMeter->oldNumOfBlocks--; pTable->oldNumOfBlocks--;
} }
while (query.over == 0) { while (query.over == 0) {
...@@ -753,17 +753,17 @@ _again: ...@@ -753,17 +753,17 @@ _again:
pCompBlock->last = 1; pCompBlock->last = 1;
if (vnodeWriteBlockToFile(pObj, pCompBlock, data, cdata, pointsRead) < 0) goto _over; if (vnodeWriteBlockToFile(pObj, pCompBlock, data, cdata, pointsRead) < 0) goto _over;
if (pCompBlock->keyLast > pObj->lastKeyOnFile) pObj->lastKeyOnFile = pCompBlock->keyLast; if (pCompBlock->keyLast > pObj->lastKeyOnFile) pObj->lastKeyOnFile = pCompBlock->keyLast;
pMeter->last = pCompBlock->last; pTable->last = pCompBlock->last;
// write block info into header buffer // write block info into header buffer
headLen += sizeof(SCompBlock); headLen += sizeof(SCompBlock);
pMeter->newNumOfBlocks++; pTable->newNumOfBlocks++;
pMeter->committedPoints += (pointsRead - pointsReadLast); pTable->committedPoints += (pointsRead - pointsReadLast);
dTrace("vid:%d sid:%d id:%s, pointsRead:%d, pointsReadLast:%d lastKey:%" PRId64 ", " dTrace("vid:%d sid:%d id:%s, pointsRead:%d, pointsReadLast:%d lastKey:%" PRId64 ", "
"slot:%d pos:%d newNumOfBlocks:%d headLen:%d", "slot:%d pos:%d newNumOfBlocks:%d headLen:%d",
pObj->vnode, pObj->sid, pObj->meterId, pointsRead, pointsReadLast, pObj->lastKeyOnFile, query.slot, query.pos, pObj->vnode, pObj->sid, pObj->meterId, pointsRead, pointsReadLast, pObj->lastKeyOnFile, query.slot, query.pos,
pMeter->newNumOfBlocks, headLen); pTable->newNumOfBlocks, headLen);
if (pointsRead < pObj->pointsPerFileBlock || query.keyIsMet) break; if (pointsRead < pObj->pointsPerFileBlock || query.keyIsMet) break;
...@@ -772,12 +772,12 @@ _again: ...@@ -772,12 +772,12 @@ _again:
} }
dTrace("vid:%d sid:%d id:%s, %d points are committed, lastKey:%" PRId64 " slot:%d pos:%d newNumOfBlocks:%d", dTrace("vid:%d sid:%d id:%s, %d points are committed, lastKey:%" PRId64 " slot:%d pos:%d newNumOfBlocks:%d",
pObj->vnode, pObj->sid, pObj->meterId, pMeter->committedPoints, pObj->lastKeyOnFile, query.slot, query.pos, pObj->vnode, pObj->sid, pObj->meterId, pTable->committedPoints, pObj->lastKeyOnFile, query.slot, query.pos,
pMeter->newNumOfBlocks); pTable->newNumOfBlocks);
if (pMeter->committedPoints > 0) { if (pTable->committedPoints > 0) {
pMeter->commitSlot = query.slot; pTable->commitSlot = query.slot;
pMeter->commitPos = query.pos; pTable->commitPos = query.pos;
} }
TSKEY nextKey = 0; TSKEY nextKey = 0;
...@@ -805,19 +805,19 @@ _again: ...@@ -805,19 +805,19 @@ _again:
continue; continue;
} }
pMeter = meterInfo + sid; pTable = meterInfo + sid;
pMeter->compInfoOffset = compInfoOffset; pTable->compInfoOffset = compInfoOffset;
pMeter->finalNumOfBlocks = pMeter->oldNumOfBlocks + pMeter->newNumOfBlocks; pTable->finalNumOfBlocks = pTable->oldNumOfBlocks + pTable->newNumOfBlocks;
if (pMeter->finalNumOfBlocks > 0) { if (pTable->finalNumOfBlocks > 0) {
pHeader->compInfoOffset = pMeter->compInfoOffset; pHeader->compInfoOffset = pTable->compInfoOffset;
compInfoOffset += sizeof(SCompInfo) + pMeter->finalNumOfBlocks * sizeof(SCompBlock) + sizeof(TSCKSUM); compInfoOffset += sizeof(SCompInfo) + pTable->finalNumOfBlocks * sizeof(SCompBlock) + sizeof(TSCKSUM);
} else { } else {
pHeader->compInfoOffset = 0; pHeader->compInfoOffset = 0;
} }
dTrace("vid:%d sid:%d id:%s, oldBlocks:%d numOfBlocks:%d compInfoOffset:%d", pObj->vnode, pObj->sid, pObj->meterId, dTrace("vid:%d sid:%d id:%s, oldBlocks:%d numOfBlocks:%d compInfoOffset:%d", pObj->vnode, pObj->sid, pObj->meterId,
pMeter->oldNumOfBlocks, pMeter->finalNumOfBlocks, compInfoOffset); pTable->oldNumOfBlocks, pTable->finalNumOfBlocks, compInfoOffset);
} }
// write the comp header into new file // write the comp header into new file
...@@ -838,16 +838,16 @@ _again: ...@@ -838,16 +838,16 @@ _again:
pObj = (SMeterObj *)(pVnode->meterList[sid]); pObj = (SMeterObj *)(pVnode->meterList[sid]);
if (pObj == NULL) continue; if (pObj == NULL) continue;
pMeter = meterInfo + sid; pTable = meterInfo + sid;
if (pMeter->finalNumOfBlocks <= 0) continue; if (pTable->finalNumOfBlocks <= 0) continue;
compInfo.last = pMeter->last; compInfo.last = pTable->last;
compInfo.uid = pObj->uid; compInfo.uid = pObj->uid;
compInfo.numOfBlocks = pMeter->finalNumOfBlocks; compInfo.numOfBlocks = pTable->finalNumOfBlocks;
/* compInfo.compBlockLen = pMeter->finalCompBlockLen; */ /* compInfo.compBlockLen = pTable->finalCompBlockLen; */
compInfo.delimiter = TSDB_VNODE_DELIMITER; compInfo.delimiter = TSDB_VNODE_DELIMITER;
taosCalcChecksumAppend(0, (uint8_t *)(&compInfo), sizeof(SCompInfo)); taosCalcChecksumAppend(0, (uint8_t *)(&compInfo), sizeof(SCompInfo));
lseek(pVnode->nfd, pMeter->compInfoOffset, SEEK_SET); lseek(pVnode->nfd, pTable->compInfoOffset, SEEK_SET);
if (twrite(pVnode->nfd, &compInfo, sizeof(compInfo)) <= 0) { if (twrite(pVnode->nfd, &compInfo, sizeof(compInfo)) <= 0) {
dError("vid:%d sid:%d id:%s, failed to write:%s, reason:%s", vnode, sid, pObj->meterId, pVnode->nfn, dError("vid:%d sid:%d id:%s, failed to write:%s, reason:%s", vnode, sid, pObj->meterId, pVnode->nfn,
strerror(errno)); strerror(errno));
...@@ -857,23 +857,23 @@ _again: ...@@ -857,23 +857,23 @@ _again:
// write the old comp blocks // write the old comp blocks
chksum = 0; chksum = 0;
if (pVnode->hfd && pMeter->oldNumOfBlocks) { if (pVnode->hfd && pTable->oldNumOfBlocks) {
lseek(pVnode->hfd, pMeter->oldCompBlockOffset, SEEK_SET); lseek(pVnode->hfd, pTable->oldCompBlockOffset, SEEK_SET);
if (pMeter->changed) { if (pTable->changed) {
int compBlockLen = pMeter->oldNumOfBlocks * sizeof(SCompBlock); int compBlockLen = pTable->oldNumOfBlocks * sizeof(SCompBlock);
read(pVnode->hfd, pOldCompBlocks, compBlockLen); read(pVnode->hfd, pOldCompBlocks, compBlockLen);
twrite(pVnode->nfd, pOldCompBlocks, compBlockLen); twrite(pVnode->nfd, pOldCompBlocks, compBlockLen);
chksum = taosCalcChecksum(0, pOldCompBlocks, compBlockLen); chksum = taosCalcChecksum(0, pOldCompBlocks, compBlockLen);
} else { } else {
tsendfile(pVnode->nfd, pVnode->hfd, NULL, pMeter->oldNumOfBlocks * sizeof(SCompBlock)); tsendfile(pVnode->nfd, pVnode->hfd, NULL, pTable->oldNumOfBlocks * sizeof(SCompBlock));
read(pVnode->hfd, &chksum, sizeof(TSCKSUM)); read(pVnode->hfd, &chksum, sizeof(TSCKSUM));
} }
} }
if (pMeter->newNumOfBlocks) { if (pTable->newNumOfBlocks) {
chksum = taosCalcChecksum(chksum, (uint8_t *)(hmem + pMeter->tempHeadOffset), chksum = taosCalcChecksum(chksum, (uint8_t *)(hmem + pTable->tempHeadOffset),
pMeter->newNumOfBlocks * sizeof(SCompBlock)); pTable->newNumOfBlocks * sizeof(SCompBlock));
if (twrite(pVnode->nfd, hmem + pMeter->tempHeadOffset, pMeter->newNumOfBlocks * sizeof(SCompBlock)) <= 0) { if (twrite(pVnode->nfd, hmem + pTable->tempHeadOffset, pTable->newNumOfBlocks * sizeof(SCompBlock)) <= 0) {
dError("vid:%d sid:%d id:%s, failed to write:%s, reason:%s", vnode, sid, pObj->meterId, pVnode->nfn, dError("vid:%d sid:%d id:%s, failed to write:%s, reason:%s", vnode, sid, pObj->meterId, pVnode->nfn,
strerror(errno)); strerror(errno));
vnodeRecoverFromPeer(pVnode, pVnode->commitFileId); vnodeRecoverFromPeer(pVnode, pVnode->commitFileId);
...@@ -891,11 +891,11 @@ _again: ...@@ -891,11 +891,11 @@ _again:
pObj = (SMeterObj *)(pVnode->meterList[sid]); pObj = (SMeterObj *)(pVnode->meterList[sid]);
if (pObj == NULL) continue; if (pObj == NULL) continue;
pMeter = meterInfo + sid; pTable = meterInfo + sid;
if (pMeter->finalNumOfBlocks <= 0) continue; if (pTable->finalNumOfBlocks <= 0) continue;
if (pMeter->committedPoints > 0) { if (pTable->committedPoints > 0) {
vnodeUpdateCommitInfo(pObj, pMeter->commitSlot, pMeter->commitPos, pMeter->commitCount); vnodeUpdateCommitInfo(pObj, pTable->commitSlot, pTable->commitPos, pTable->commitCount);
} }
} }
......
...@@ -1236,12 +1236,12 @@ _error_merge: ...@@ -1236,12 +1236,12 @@ _error_merge:
} \ } \
} }
int isCacheEnd(SBlockIter iter, SMeterObj *pMeter) { int isCacheEnd(SBlockIter iter, SMeterObj *pTable) {
SCacheInfo *pInfo = (SCacheInfo *)(pMeter->pCache); SCacheInfo *pInfo = (SCacheInfo *)(pTable->pCache);
int slot = 0; int slot = 0;
int pos = 0; int pos = 0;
if (pInfo->cacheBlocks[pInfo->currentSlot]->numOfPoints == pMeter->pointsPerBlock) { if (pInfo->cacheBlocks[pInfo->currentSlot]->numOfPoints == pTable->pointsPerBlock) {
slot = (pInfo->currentSlot + 1) % (pInfo->maxBlocks); slot = (pInfo->currentSlot + 1) % (pInfo->maxBlocks);
pos = 0; pos = 0;
} else { } else {
......
...@@ -4506,7 +4506,7 @@ int32_t vnodeMultiMeterQueryPrepare(SQInfo *pQInfo, SQuery *pQuery, void *param) ...@@ -4506,7 +4506,7 @@ int32_t vnodeMultiMeterQueryPrepare(SQInfo *pQInfo, SQuery *pQuery, void *param)
} }
// get one queried meter // get one queried meter
SMeterObj *pMeter = getMeterObj(pSupporter->pMetersHashTable, pSupporter->pSidSet->pSids[0]->sid); SMeterObj *pTable = getMeterObj(pSupporter->pMetersHashTable, pSupporter->pSidSet->pSids[0]->sid);
pRuntimeEnv->pTSBuf = param; pRuntimeEnv->pTSBuf = param;
pRuntimeEnv->cur.vnodeIndex = -1; pRuntimeEnv->cur.vnodeIndex = -1;
...@@ -4517,18 +4517,18 @@ int32_t vnodeMultiMeterQueryPrepare(SQInfo *pQInfo, SQuery *pQuery, void *param) ...@@ -4517,18 +4517,18 @@ int32_t vnodeMultiMeterQueryPrepare(SQInfo *pQInfo, SQuery *pQuery, void *param)
tsBufSetTraverseOrder(pRuntimeEnv->pTSBuf, order); tsBufSetTraverseOrder(pRuntimeEnv->pTSBuf, order);
} }
int32_t ret = setupQueryRuntimeEnv(pMeter, pQuery, &pSupporter->runtimeEnv, pTagSchema, TSQL_SO_ASC, true); int32_t ret = setupQueryRuntimeEnv(pTable, pQuery, &pSupporter->runtimeEnv, pTagSchema, TSQL_SO_ASC, true);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
return ret; return ret;
} }
ret = allocateRuntimeEnvBuf(pRuntimeEnv, pMeter); ret = allocateRuntimeEnvBuf(pRuntimeEnv, pTable);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
return ret; return ret;
} }
tSidSetSort(pSupporter->pSidSet); tSidSetSort(pSupporter->pSidSet);
vnodeRecordAllFiles(pQInfo, pMeter->vnode); vnodeRecordAllFiles(pQInfo, pTable->vnode);
if ((ret = allocateOutputBufForGroup(pSupporter, pQuery, true)) != TSDB_CODE_SUCCESS) { if ((ret = allocateOutputBufForGroup(pSupporter, pQuery, true)) != TSDB_CODE_SUCCESS) {
return ret; return ret;
...@@ -4595,12 +4595,12 @@ void vnodeDecMeterRefcnt(SQInfo *pQInfo) { ...@@ -4595,12 +4595,12 @@ void vnodeDecMeterRefcnt(SQInfo *pQInfo) {
} else { } else {
int32_t num = 0; int32_t num = 0;
for (int32_t i = 0; i < pSupporter->numOfMeters; ++i) { for (int32_t i = 0; i < pSupporter->numOfMeters; ++i) {
SMeterObj *pMeter = getMeterObj(pSupporter->pMetersHashTable, pSupporter->pSidSet->pSids[i]->sid); SMeterObj *pTable = getMeterObj(pSupporter->pMetersHashTable, pSupporter->pSidSet->pSids[i]->sid);
atomic_fetch_sub_32(&(pMeter->numOfQueries), 1); atomic_fetch_sub_32(&(pTable->numOfQueries), 1);
if (pMeter->numOfQueries > 0) { if (pTable->numOfQueries > 0) {
dTrace("QInfo:%p vid:%d sid:%d meterId:%s, query is over, numOfQueries:%d", pQInfo, pMeter->vnode, pMeter->sid, dTrace("QInfo:%p vid:%d sid:%d meterId:%s, query is over, numOfQueries:%d", pQInfo, pTable->vnode, pTable->sid,
pMeter->meterId, pMeter->numOfQueries); pTable->meterId, pTable->numOfQueries);
num++; num++;
} }
} }
......
...@@ -186,15 +186,15 @@ void vnodeUpdateStreamRole(SVnodeObj *pVnode) { ...@@ -186,15 +186,15 @@ void vnodeUpdateStreamRole(SVnodeObj *pVnode) {
// Callback function called from client // Callback function called from client
void vnodeCloseStreamCallback(void *param) { void vnodeCloseStreamCallback(void *param) {
SMeterObj *pMeter = (SMeterObj *)param; SMeterObj *pTable = (SMeterObj *)param;
SVnodeObj *pVnode = NULL; SVnodeObj *pVnode = NULL;
if (pMeter == NULL || pMeter->sqlLen == 0) return; if (pTable == NULL || pTable->sqlLen == 0) return;
pVnode = vnodeList + pMeter->vnode; pVnode = vnodeList + pTable->vnode;
pMeter->sqlLen = 0; pTable->sqlLen = 0;
pMeter->pSql = NULL; pTable->pSql = NULL;
pMeter->pStream = NULL; pTable->pStream = NULL;
pVnode->numOfStreams--; pVnode->numOfStreams--;
...@@ -203,5 +203,5 @@ void vnodeCloseStreamCallback(void *param) { ...@@ -203,5 +203,5 @@ void vnodeCloseStreamCallback(void *param) {
pVnode->dbConn = NULL; pVnode->dbConn = NULL;
} }
vnodeSaveMeterObjToFile(pMeter); vnodeSaveMeterObjToFile(pTable);
} }
\ No newline at end of file
...@@ -527,7 +527,7 @@ bool vnodeIsProjectionQuery(SSqlFunctionExpr* pExpr, int32_t numOfOutput) { ...@@ -527,7 +527,7 @@ bool vnodeIsProjectionQuery(SSqlFunctionExpr* pExpr, int32_t numOfOutput) {
} }
/* /*
* the pMeter->state may be changed by vnodeIsSafeToDeleteMeter and import/update processor, the check of * the pTable->state may be changed by vnodeIsSafeToDeleteMeter and import/update processor, the check of
* the state will not always be correct. * the state will not always be correct.
* *
* The import/update/deleting is actually blocked by current query processing if the check of meter state is * The import/update/deleting is actually blocked by current query processing if the check of meter state is
...@@ -548,30 +548,30 @@ int32_t vnodeIncQueryRefCount(SQueryMeterMsg* pQueryMsg, SMeterSidExtInfo** pSid ...@@ -548,30 +548,30 @@ int32_t vnodeIncQueryRefCount(SQueryMeterMsg* pQueryMsg, SMeterSidExtInfo** pSid
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < pQueryMsg->numOfSids; ++i) { for (int32_t i = 0; i < pQueryMsg->numOfSids; ++i) {
SMeterObj* pMeter = pVnode->meterList[pSids[i]->sid]; SMeterObj* pTable = pVnode->meterList[pSids[i]->sid];
/* /*
* If table is missing or is in dropping status, config it from management node, and ignore it * If table is missing or is in dropping status, config it from management node, and ignore it
* during query processing. The error code of TSDB_CODE_NOT_ACTIVE_TABLE will never return to client. * during query processing. The error code of TSDB_CODE_NOT_ACTIVE_TABLE will never return to client.
* The missing table needs to be removed from pSids list * The missing table needs to be removed from pSids list
*/ */
if (pMeter == NULL || vnodeIsMeterState(pMeter, TSDB_METER_STATE_DROPPING)) { if (pTable == NULL || vnodeIsMeterState(pTable, TSDB_METER_STATE_DROPPING)) {
dWarn("qmsg:%p, vid:%d sid:%d, not there or will be dropped, ignore this table in query", pQueryMsg, dWarn("qmsg:%p, vid:%d sid:%d, not there or will be dropped, ignore this table in query", pQueryMsg,
pQueryMsg->vnode, pSids[i]->sid); pQueryMsg->vnode, pSids[i]->sid);
vnodeSendMeterCfgMsg(pQueryMsg->vnode, pSids[i]->sid); vnodeSendMeterCfgMsg(pQueryMsg->vnode, pSids[i]->sid);
continue; continue;
} else if (pMeter->uid != pSids[i]->uid || pMeter->sid != pSids[i]->sid) { } else if (pTable->uid != pSids[i]->uid || pTable->sid != pSids[i]->sid) {
code = TSDB_CODE_TABLE_ID_MISMATCH; code = TSDB_CODE_TABLE_ID_MISMATCH;
dError("qmsg:%p, vid:%d sid:%d id:%s uid:%" PRIu64 ", id mismatch. sid:%d uid:%" PRId64 " in msg", pQueryMsg, dError("qmsg:%p, vid:%d sid:%d id:%s uid:%" PRIu64 ", id mismatch. sid:%d uid:%" PRId64 " in msg", pQueryMsg,
pQueryMsg->vnode, pMeter->sid, pMeter->meterId, pMeter->uid, pSids[i]->sid, pSids[i]->uid); pQueryMsg->vnode, pTable->sid, pTable->meterId, pTable->uid, pSids[i]->sid, pSids[i]->uid);
vnodeSendMeterCfgMsg(pQueryMsg->vnode, pSids[i]->sid); vnodeSendMeterCfgMsg(pQueryMsg->vnode, pSids[i]->sid);
continue; continue;
} else if (pMeter->state > TSDB_METER_STATE_INSERTING) { //update or import } else if (pTable->state > TSDB_METER_STATE_INSERTING) { //update or import
code = TSDB_CODE_ACTION_IN_PROGRESS; code = TSDB_CODE_ACTION_IN_PROGRESS;
dTrace("qmsg:%p, vid:%d sid:%d id:%s, it is in state:%s, wait!", pQueryMsg, pQueryMsg->vnode, pSids[i]->sid, dTrace("qmsg:%p, vid:%d sid:%d id:%s, it is in state:%s, wait!", pQueryMsg, pQueryMsg->vnode, pSids[i]->sid,
pMeter->meterId, taosGetTableStatusStr(pMeter->state)); pTable->meterId, taosGetTableStatusStr(pTable->state));
continue; continue;
} }
...@@ -579,15 +579,15 @@ int32_t vnodeIncQueryRefCount(SQueryMeterMsg* pQueryMsg, SMeterSidExtInfo** pSid ...@@ -579,15 +579,15 @@ int32_t vnodeIncQueryRefCount(SQueryMeterMsg* pQueryMsg, SMeterSidExtInfo** pSid
* vnodeIsSafeToDeleteMeter will wait for this function complete, and then it can * vnodeIsSafeToDeleteMeter will wait for this function complete, and then it can
* check if the numOfQueries is 0 or not. * check if the numOfQueries is 0 or not.
*/ */
pMeterObjList[(*numOfIncTables)++] = pMeter; pMeterObjList[(*numOfIncTables)++] = pTable;
atomic_fetch_add_32(&pMeter->numOfQueries, 1); atomic_fetch_add_32(&pTable->numOfQueries, 1);
pSids[index++] = pSids[i]; pSids[index++] = pSids[i];
// output for meter more than one query executed // output for meter more than one query executed
if (pMeter->numOfQueries > 1) { if (pTable->numOfQueries > 1) {
dTrace("qmsg:%p, vid:%d sid:%d id:%s, inc query ref, numOfQueries:%d", pQueryMsg, pMeter->vnode, pMeter->sid, dTrace("qmsg:%p, vid:%d sid:%d id:%s, inc query ref, numOfQueries:%d", pQueryMsg, pTable->vnode, pTable->sid,
pMeter->meterId, pMeter->numOfQueries); pTable->meterId, pTable->numOfQueries);
num++; num++;
} }
} }
...@@ -605,14 +605,14 @@ void vnodeDecQueryRefCount(SQueryMeterMsg* pQueryMsg, SMeterObj** pMeterObjList, ...@@ -605,14 +605,14 @@ void vnodeDecQueryRefCount(SQueryMeterMsg* pQueryMsg, SMeterObj** pMeterObjList,
int32_t num = 0; int32_t num = 0;
for (int32_t i = 0; i < numOfIncTables; ++i) { for (int32_t i = 0; i < numOfIncTables; ++i) {
SMeterObj* pMeter = pMeterObjList[i]; SMeterObj* pTable = pMeterObjList[i];
if (pMeter != NULL) { // here, do not need to lock to perform operations if (pTable != NULL) { // here, do not need to lock to perform operations
atomic_fetch_sub_32(&pMeter->numOfQueries, 1); atomic_fetch_sub_32(&pTable->numOfQueries, 1);
if (pMeter->numOfQueries > 0) { if (pTable->numOfQueries > 0) {
dTrace("qmsg:%p, vid:%d sid:%d id:%s dec query ref, numOfQueries:%d", pQueryMsg, pMeter->vnode, pMeter->sid, dTrace("qmsg:%p, vid:%d sid:%d id:%s dec query ref, numOfQueries:%d", pQueryMsg, pTable->vnode, pTable->sid,
pMeter->meterId, pMeter->numOfQueries); pTable->meterId, pTable->numOfQueries);
num++; num++;
} }
} }
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册