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

rename meter to table

This commit is contained in:
slguan 2020-02-24 14:48:14 +08:00
parent 03da5549ce
commit 7cedf4afd6
49 changed files with 303 additions and 335 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/client/inc/tscUtil.h
View File

@ -67,7 +67,7 @@ typedef struct SJoinSubquerySupporter {
} SJoinSubquerySupporter;
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
SMeterMeta* pMeterMeta, STableDataBlocks** dataBlocks);
STableMeta* pMeterMeta, STableDataBlocks** dataBlocks);
void tscAppendDataBlock(SDataBlockList* pList, STableDataBlocks* pBlocks);
void tscDestroyDataBlock(STableDataBlocks* pDataBlock);
@ -81,11 +81,11 @@ int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock);
void tscFreeUnusedDataBlocks(SDataBlockList* pList);
int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SDataBlockList* pDataList);
int32_t tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList, int64_t id, int32_t size,
int32_t startOffset, int32_t rowSize, const char* tableId, SMeterMeta* pMeterMeta,
int32_t startOffset, int32_t rowSize, const char* tableId, STableMeta* pMeterMeta,
STableDataBlocks** dataBlocks);
SVnodeSidList* tscGetVnodeSidList(SMetricMeta* pMetricmeta, int32_t vnodeIdx);
SMeterSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);
SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx);
STableSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);
/**
*
@ -120,7 +120,7 @@ void tscClearInterpInfo(SQueryInfo* pQueryInfo);
bool tscIsInsertOrImportData(char* sqlstr);
/* use for keep current db info temporarily, for handle table with db prefix */
void tscGetDBInfoFromMeterId(char* meterId, char* db);
void tscGetDBInfoFromMeterId(char* tableId, char* db);
int tscAllocPayload(SSqlCmd* pCmd, int size);
@ -196,7 +196,7 @@ int32_t tscGetQueryInfoDetailSafely(SSqlCmd *pCmd, int32_t subClauseIndex, SQuer
SMeterMetaInfo* tscGetMeterMetaInfoByUid(SQueryInfo* pQueryInfo, uint64_t uid, int32_t* index);
void tscClearMeterMetaInfo(SMeterMetaInfo* pMeterMetaInfo, bool removeFromCache);
SMeterMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, SMeterMeta* pMeterMeta, SMetricMeta* pMetricMeta,
SMeterMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, STableMeta* pMeterMeta, SSuperTableMeta* pMetricMeta,
int16_t numOfTags, int16_t* tags);
SMeterMetaInfo* tscAddEmptyMeterMetaInfo(SQueryInfo *pQueryInfo);
int32_t tscAddSubqueryInfo(SSqlCmd *pCmd);

8
src/client/inc/tsclient.h
View File

@ -47,8 +47,8 @@ typedef struct SSqlGroupbyExpr {
} SSqlGroupbyExpr;
typedef struct SMeterMetaInfo {
SMeterMeta * pMeterMeta; // metermeta
SMetricMeta *pMetricMeta; // metricmeta
STableMeta * pMeterMeta; // metermeta
SSuperTableMeta *pMetricMeta; // metricmeta
/*
* 1. keep the vnode index during the multi-vnode super table projection query
@ -160,7 +160,7 @@ typedef struct STableDataBlocks {
bool ordered; // if current rows are ordered or not
int64_t vgid; // virtual group id
int64_t prevTS; // previous timestamp, recorded to decide if the records array is ts ascending
int32_t numOfMeters; // number of tables in current submit block
int32_t numOfTables; // number of tables in current submit block
int32_t rowSize; // row size for current table
uint32_t nAllocSize;
@ -171,7 +171,7 @@ typedef struct STableDataBlocks {
* the metermeta for current table, the metermeta will be used during submit stage, keep a ref
* to avoid it to be removed from cache
*/
SMeterMeta *pMeterMeta;
STableMeta *pMeterMeta;
union {
char *filename;

6
src/client/src/tscAsync.c
View File

@ -444,10 +444,10 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
if (code != 0) {
code = abs(code);
pRes->code = code;
tscTrace("%p failed to renew meterMeta", pSql);
tscTrace("%p failed to renew tableMeta", pSql);
tsem_post(&pSql->rspSem);
} else {
tscTrace("%p renew meterMeta successfully, command:%d, code:%d, retry:%d",
tscTrace("%p renew tableMeta successfully, command:%d, code:%d, retry:%d",
pSql, pSql->cmd.command, pSql->res.code, pSql->retry);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, 0, 0);
@ -546,7 +546,7 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
tscTansformSQLFunctionForSTableQuery(pQueryInfo);
tscIncStreamExecutionCount(pSql->pStream);
} else {
tscTrace("%p get meterMeta/metricMeta successfully", pSql);
tscTrace("%p get tableMeta/metricMeta successfully", pSql);
}
tscDoQuery(pSql);

16
src/client/src/tscLocal.c
View File

@ -77,7 +77,7 @@ static int32_t getToStringLength(const char *pData, int32_t length, int32_t type
* length((uint64_t) 123456789011) > 12, greater than sizsof(uint64_t)
*/
static int32_t tscMaxLengthOfTagsFields(SSqlObj *pSql) {
SMeterMeta *pMeta = tscGetMeterMetaInfo(&pSql->cmd, 0, 0)->pMeterMeta;
STableMeta *pMeta = tscGetMeterMetaInfo(&pSql->cmd, 0, 0)->pMeterMeta;
if (pMeta->tableType == TSDB_TABLE_TYPE_SUPER_TABLE || pMeta->tableType == TSDB_TABLE_TYPE_NORMAL_TABLE ||
pMeta->tableType == TSDB_TABLE_TYPE_STREAM_TABLE) {
@ -109,7 +109,7 @@ static int32_t tscSetValueToResObj(SSqlObj *pSql, int32_t rowLen) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
SMeterMeta * pMeta = pMeterMetaInfo->pMeterMeta;
STableMeta * pMeta = pMeterMetaInfo->pMeterMeta;
/*
* tagValueCnt is to denote the number of tags columns for meter, not metric. and is to show the column data.
@ -285,7 +285,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
SMetricMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
SSuperTableMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
SSchema * pSchema = tsGetTagSchema(pMeterMetaInfo->pMeterMeta);
int32_t vOffset[TSDB_MAX_COLUMNS] = {0};
@ -299,7 +299,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
}
}
int32_t totalNumOfResults = pMetricMeta->numOfMeters;
int32_t totalNumOfResults = pMetricMeta->numOfTables;
int32_t rowLen = tscGetResRowLength(pQueryInfo);
tscInitResObjForLocalQuery(pSql, totalNumOfResults, rowLen);
@ -309,7 +309,7 @@ static int tscBuildMetricTagProjectionResult(SSqlObj *pSql) {
SVnodeSidList *pSidList = (SVnodeSidList *)((char *)pMetricMeta + pMetricMeta->list[i]);
for (int32_t j = 0; j < pSidList->numOfSids; ++j) {
SMeterSidExtInfo *pSidExt = tscGetMeterSidInfo(pSidList, j);
STableSidExtInfo *pSidExt = tscGetMeterSidInfo(pSidList, j);
for (int32_t k = 0; k < pQueryInfo->fieldsInfo.numOfOutputCols; ++k) {
SColIndexEx *pColIndex = &tscSqlExprGet(pQueryInfo, k)->colInfo;
@ -336,7 +336,7 @@ static int tscBuildMetricTagSqlFunctionResult(SSqlObj *pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
SMetricMeta *pMetricMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0)->pMetricMeta;
SSuperTableMeta *pMetricMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0)->pMetricMeta;
int32_t totalNumOfResults = 1; // count function only produce one result
int32_t rowLen = tscGetResRowLength(pQueryInfo);
@ -351,7 +351,7 @@ static int tscBuildMetricTagSqlFunctionResult(SSqlObj *pSql) {
TAOS_FIELD *pField = tscFieldInfoGetField(pQueryInfo, k);
memcpy(pRes->data + tscFieldInfoGetOffset(pQueryInfo, i) * totalNumOfResults + pField->bytes * rowIdx,
&pMetricMeta->numOfMeters, sizeof(pMetricMeta->numOfMeters));
&pMetricMeta->numOfTables, sizeof(pMetricMeta->numOfTables));
} else {
tscError("not support operations");
continue;
@ -368,7 +368,7 @@ static int tscProcessQueryTags(SSqlObj *pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
SMeterMeta *pMeterMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0)->pMeterMeta;
STableMeta *pMeterMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0)->pMeterMeta;
if (pMeterMeta == NULL || pMeterMeta->numOfTags == 0 || pMeterMeta->numOfColumns == 0) {
strcpy(pCmd->payload, "invalid table");
pSql->res.code = TSDB_CODE_INVALID_TABLE;

16
src/client/src/tscParseInsert.c
View File

@ -496,7 +496,7 @@ static int32_t rowDataCompar(const void *lhs, const void *rhs) {
}
}
int tsParseValues(char **str, STableDataBlocks *pDataBlock, SMeterMeta *pMeterMeta, int maxRows,
int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pMeterMeta, int maxRows,
SParsedDataColInfo *spd, char *error, int32_t *code, char *tmpTokenBuf) {
int32_t index = 0;
SSQLToken sToken;
@ -601,7 +601,7 @@ int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int3
return TSDB_CODE_SUCCESS;
}
static void tsSetBlockInfo(SShellSubmitBlock *pBlocks, const SMeterMeta *pMeterMeta, int32_t numOfRows) {
static void tsSetBlockInfo(SShellSubmitBlock *pBlocks, const STableMeta *pMeterMeta, int32_t numOfRows) {
pBlocks->sid = pMeterMeta->sid;
pBlocks->uid = pMeterMeta->uid;
pBlocks->sversion = pMeterMeta->sversion;
@ -655,7 +655,7 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableHashList, char
int32_t *totalNum) {
SSqlCmd * pCmd = &pSql->cmd;
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0);
SMeterMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableDataBlocks *dataBuf = NULL;
int32_t ret = tscGetDataBlockFromList(pTableHashList, pCmd->pDataBlocks, pMeterMeta->uid, TSDB_DEFAULT_PAYLOAD_SIZE,
@ -695,7 +695,7 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableHashList, char
tsSetBlockInfo(pBlocks, pMeterMeta, numOfRows);
dataBuf->vgid = pMeterMeta->vgid;
dataBuf->numOfMeters = 1;
dataBuf->numOfTables = 1;
/*
* the value of pRes->numOfRows does not affect the true result of AFFECTED ROWS,
@ -1136,7 +1136,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
wordfree(&full_path);
STableDataBlocks *pDataBlock = NULL;
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
int32_t ret = tscCreateDataBlock(PATH_MAX, pMeterMeta->rowSize, sizeof(SShellSubmitBlock), pMeterMetaInfo->name,
pMeterMeta, &pDataBlock);
@ -1148,7 +1148,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
strcpy(pDataBlock->filename, fname);
} else if (sToken.type == TK_LP) {
/* insert into tablename(col1, col2,..., coln) values(v1, v2,... vn); */
SMeterMeta *pMeterMeta = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0)->pMeterMeta;
STableMeta *pMeterMeta = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0)->pMeterMeta;
SSchema * pSchema = tsGetSchema(pMeterMeta);
if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
@ -1349,7 +1349,7 @@ static int doPackSendDataBlock(SSqlObj *pSql, int32_t numOfRows, STableDataBlock
SSqlCmd *pCmd = &pSql->cmd;
assert(pCmd->numOfClause == 1);
SMeterMeta *pMeterMeta = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0)->pMeterMeta;
STableMeta *pMeterMeta = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0)->pMeterMeta;
SShellSubmitBlock *pBlocks = (SShellSubmitBlock *)(pTableDataBlocks->pData);
tsSetBlockInfo(pBlocks, pMeterMeta, numOfRows);
@ -1383,7 +1383,7 @@ static int tscInsertDataFromFile(SSqlObj *pSql, FILE *fp, char *tmpTokenBuf) {
int nrows = 0;
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0);
SMeterMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
assert(pCmd->numOfClause == 1);
int32_t rowSize = pMeterMeta->rowSize;

32
src/client/src/tscSQLParser.c
View File

@ -883,7 +883,7 @@ bool validateOneTags(SSqlCmd* pCmd, TAOS_FIELD* pTagField) {
assert(pCmd->numOfClause == 1);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
// no more than 6 tags
if (pMeterMeta->numOfTags == TSDB_MAX_TAGS) {
@ -953,7 +953,7 @@ bool validateOneColumn(SSqlCmd* pCmd, TAOS_FIELD* pColField) {
assert(pCmd->numOfClause == 1);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
// no more max columns
if (pMeterMeta->numOfColumns >= TSDB_MAX_COLUMNS ||
@ -1215,7 +1215,7 @@ int32_t insertResultField(SQueryInfo* pQueryInfo, int32_t outputIndex, SColumnLi
SSqlExpr* doAddProjectCol(SQueryInfo* pQueryInfo, int32_t outputIndex, int32_t colIdx, int32_t tableIndex) {
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, tableIndex);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
SSchema* pSchema = tsGetColumnSchema(pMeterMeta, colIdx);
int32_t numOfCols = pMeterMeta->numOfColumns;
@ -1267,7 +1267,7 @@ static void addProjectQueryCol(SQueryInfo* pQueryInfo, int32_t startPos, SColumn
SSqlExpr* pExpr = doAddProjectCol(pQueryInfo, startPos, pIndex->columnIndex, pIndex->tableIndex);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, pIndex->tableIndex);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
SSchema* pSchema = tsGetColumnSchema(pMeterMeta, pIndex->columnIndex);
@ -1306,7 +1306,7 @@ static int32_t doAddProjectionExprAndResultFields(SQueryInfo* pQueryInfo, SColum
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, pIndex->tableIndex);
int32_t numOfTotalColumns = 0;
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
SSchema* pSchema = tsGetSchema(pMeterMeta);
if (UTIL_METER_IS_SUPERTABLE(pMeterMetaInfo)) {
@ -1368,7 +1368,7 @@ int32_t addProjectionExprAndResultField(SQueryInfo* pQueryInfo, tSQLExprItem* pI
tscAddSpecialColumnForSelect(pQueryInfo, startPos, TSDB_FUNC_TAGPRJ, &index, &colSchema, true);
} else {
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index.tableIndex);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
if (index.columnIndex >= pMeterMeta->numOfColumns && UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) {
return invalidSqlErrMsg(pQueryInfo->msg, msg1);
@ -1835,7 +1835,7 @@ static bool isTablenameToken(SSQLToken* token) {
}
static int16_t doGetColumnIndex(SQueryInfo* pQueryInfo, int32_t index, SSQLToken* pToken) {
SMeterMeta* pMeterMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index)->pMeterMeta;
STableMeta* pMeterMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index)->pMeterMeta;
int32_t numOfCols = pMeterMeta->numOfColumns + pMeterMeta->numOfTags;
SSchema* pSchema = tsGetSchema(pMeterMeta);
@ -2360,7 +2360,7 @@ int32_t parseGroupbyClause(SQueryInfo* pQueryInfo, tVariantList* pList, SSqlCmd*
return invalidSqlErrMsg(pQueryInfo->msg, msg1);
}
SMeterMeta* pMeterMeta = NULL;
STableMeta* pMeterMeta = NULL;
SSchema* pSchema = NULL;
SSchema s = tsGetTbnameColumnSchema();
@ -2704,7 +2704,7 @@ enum {
static int32_t extractColumnFilterInfo(SQueryInfo* pQueryInfo, SColumnIndex* pIndex, tSQLExpr* pExpr, int32_t sqlOptr) {
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, pIndex->tableIndex);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
SSchema* pSchema = tsGetColumnSchema(pMeterMeta, pIndex->columnIndex);
const char* msg1 = "non binary column not support like operator";
@ -2927,7 +2927,7 @@ static int32_t validateSQLExpr(tSQLExpr* pExpr, SQueryInfo* pQueryInfo, SColumnL
}
// if column is timestamp, bool, binary, nchar, not support arithmetic, so return invalid sql
SMeterMeta* pMeterMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index.tableIndex)->pMeterMeta;
STableMeta* pMeterMeta = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index.tableIndex)->pMeterMeta;
SSchema* pSchema = tsGetSchema(pMeterMeta) + index.columnIndex;
if ((pSchema->type == TSDB_DATA_TYPE_TIMESTAMP) || (pSchema->type == TSDB_DATA_TYPE_BOOL)
|| (pSchema->type == TSDB_DATA_TYPE_BINARY) || (pSchema->type == TSDB_DATA_TYPE_NCHAR)){
@ -3140,7 +3140,7 @@ static int32_t handleExprInQueryCond(SQueryInfo* pQueryInfo, tSQLExpr** pExpr, S
assert(isExprDirectParentOfLeaftNode(*pExpr));
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index.tableIndex);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
if (index.columnIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX) { // query on time range
if (!validateJoinExprNode(pQueryInfo, *pExpr, &index)) {
@ -3485,7 +3485,7 @@ static int32_t getTimeRangeFromExpr(SQueryInfo* pQueryInfo, tSQLExpr* pExpr) {
}
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, index.tableIndex);
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
tSQLExpr* pRight = pExpr->pRight;
@ -4086,7 +4086,7 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
return ret;
}
SMeterMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta* pMeterMeta = pMeterMetaInfo->pMeterMeta;
if (pAlterSQL->type == TSDB_ALTER_TABLE_ADD_TAG_COLUMN || pAlterSQL->type == TSDB_ALTER_TABLE_DROP_TAG_COLUMN ||
pAlterSQL->type == TSDB_ALTER_TABLE_CHANGE_TAG_COLUMN) {
@ -4534,8 +4534,8 @@ int32_t parseLimitClause(SQueryInfo* pQueryInfo, int32_t clauseIndex, SQuerySQL*
}
// No tables included. No results generated. Query results are empty.
SMetricMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta;
if (pMeterMetaInfo->pMeterMeta == NULL || pMetricMeta == NULL || pMetricMeta->numOfMeters == 0) {
SSuperTableMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta;
if (pMeterMetaInfo->pMeterMeta == NULL || pMetricMeta == NULL || pMetricMeta->numOfTables == 0) {
tscTrace("%p no table in metricmeta, no output result", pSql);
pQueryInfo->command = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
}
@ -4629,7 +4629,7 @@ static int32_t setTimePrecisionOption(SSqlCmd* pCmd, SCreateDbMsg* pMsg, SCreate
}
static void setCreateDBOption(SCreateDbMsg* pMsg, SCreateDBInfo* pCreateDb) {
pMsg->blocksPerMeter = htons(pCreateDb->numOfBlocksPerTable);
pMsg->blocksPerTable = htons(pCreateDb->numOfBlocksPerTable);
pMsg->compression = pCreateDb->compressionLevel;
pMsg->commitLog = (char)pCreateDb->commitLog;

28
src/client/src/tscSchemaUtil.c
View File

@ -64,14 +64,14 @@ bool isValidSchema(struct SSchema* pSchema, int32_t numOfCols) {
return (rowLen <= TSDB_MAX_BYTES_PER_ROW);
}
struct SSchema* tsGetSchema(SMeterMeta* pMeta) {
struct SSchema* tsGetSchema(STableMeta* pMeta) {
if (pMeta == NULL) {
return NULL;
}
return tsGetColumnSchema(pMeta, 0);
}
struct SSchema* tsGetTagSchema(SMeterMeta* pMeta) {
struct SSchema* tsGetTagSchema(STableMeta* pMeta) {
if (pMeta == NULL || pMeta->numOfTags == 0) {
return NULL;
}
@ -79,8 +79,8 @@ struct SSchema* tsGetTagSchema(SMeterMeta* pMeta) {
return tsGetColumnSchema(pMeta, pMeta->numOfColumns);
}
struct SSchema* tsGetColumnSchema(SMeterMeta* pMeta, int32_t startCol) {
return (SSchema*)(((char*)pMeta + sizeof(SMeterMeta)) + startCol * sizeof(SSchema));
struct SSchema* tsGetColumnSchema(STableMeta* pMeta, int32_t startCol) {
return (SSchema*)(((char*)pMeta + sizeof(STableMeta)) + startCol * sizeof(SSchema));
}
struct SSchema tsGetTbnameColumnSchema() {
@ -94,7 +94,7 @@ struct SSchema tsGetTbnameColumnSchema() {
* the MeterMeta data format in memory is as follows:
*
* +--------------------+
* |SMeterMeta Body data| sizeof(SMeterMeta)
* |STableMeta Body data| sizeof(STableMeta)
* +--------------------+
* |Schema data | numOfTotalColumns * sizeof(SSchema)
* +--------------------+
@ -104,14 +104,14 @@ struct SSchema tsGetTbnameColumnSchema() {
* @param pMeta
* @return
*/
char* tsGetTagsValue(SMeterMeta* pMeta) {
char* tsGetTagsValue(STableMeta* pMeta) {
int32_t numOfTotalCols = pMeta->numOfColumns + pMeta->numOfTags;
uint32_t offset = sizeof(SMeterMeta) + numOfTotalCols * sizeof(SSchema);
uint32_t offset = sizeof(STableMeta) + numOfTotalCols * sizeof(SSchema);
return ((char*)pMeta + offset);
}
bool tsMeterMetaIdentical(SMeterMeta* p1, SMeterMeta* p2) {
bool tsMeterMetaIdentical(STableMeta* p1, STableMeta* p2) {
if (p1 == NULL || p2 == NULL || p1->uid != p2->uid || p1->sversion != p2->sversion) {
return false;
}
@ -120,7 +120,7 @@ bool tsMeterMetaIdentical(SMeterMeta* p1, SMeterMeta* p2) {
return true;
}
size_t size = sizeof(SMeterMeta) + p1->numOfColumns * sizeof(SSchema);
size_t size = sizeof(STableMeta) + p1->numOfColumns * sizeof(SSchema);
for (int32_t i = 0; i < p1->numOfTags; ++i) {
SSchema* pColSchema = tsGetColumnSchema(p1, i + p1->numOfColumns);
@ -151,16 +151,16 @@ static FORCE_INLINE size_t copy(char* dst, const char* src, char delimiter) {
/**
* extract table name from meterid, which the format of userid.dbname.metername
* @param meterId
* @param tableId
* @return
*/
void extractTableName(char* meterId, char* name) {
char* r = skipSegments(meterId, TS_PATH_DELIMITER[0], 2);
void extractTableName(char* tableId, char* name) {
char* r = skipSegments(tableId, TS_PATH_DELIMITER[0], 2);
copy(name, r, TS_PATH_DELIMITER[0]);
}
SSQLToken extractDBName(char* meterId, char* name) {
char* r = skipSegments(meterId, TS_PATH_DELIMITER[0], 1);
SSQLToken extractDBName(char* tableId, char* name) {
char* r = skipSegments(tableId, TS_PATH_DELIMITER[0], 1);
size_t len = copy(name, r, TS_PATH_DELIMITER[0]);
SSQLToken token = {.z = name, .n = len, .type = TK_STRING};

150
src/client/src/tscServer.c
View File

@ -1215,7 +1215,7 @@ void tscUpdateVnodeInSubmitMsg(SSqlObj *pSql, char *buf) {
char * pMsg;
SMeterMetaInfo * pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, pSql->cmd.clauseIndex, 0);
SMeterMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
pMsg = buf + tsRpcHeadSize;
@ -1233,7 +1233,7 @@ int tscBuildSubmitMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
SMeterMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
pStart = pSql->cmd.payload + tsRpcHeadSize;
pMsg = pStart;
@ -1258,13 +1258,13 @@ void tscUpdateVnodeInQueryMsg(SSqlObj *pSql, char *buf) {
// SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0);
//
// char * pStart = buf + tsRpcHeadSize;
// SQueryMeterMsg *pQueryMsg = (SQueryMeterMsg *)pStart;
// SQueryTableMsg *pQueryMsg = (SQueryTableMsg *)pStart;
//
// if (UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) { // pColumnModel == NULL, query on meter
// SMeterMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
// STableMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
// pQueryMsg->vnode = htons(pMeterMeta->vpeerDesc[pSql->index].vnode);
// } else { // query on metric
// SMetricMeta * pMetricMeta = pMeterMetaInfo->pMetricMeta;
// SSuperTableMeta * pMetricMeta = pMeterMetaInfo->pMetricMeta;
// SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pMeterMetaInfo->vnodeIndex);
// pQueryMsg->vnode = htons(pVnodeSidList->vpeerDesc[pSql->index].vnode);
// }
@ -1285,14 +1285,14 @@ static int32_t tscEstimateQueryMsgSize(SSqlCmd *pCmd, int32_t clauseIndex) {
// meter query without tags values
if (!UTIL_METER_IS_SUPERTABLE(pMeterMetaInfo)) {
return MIN_QUERY_MSG_PKT_SIZE + minMsgSize() + sizeof(SQueryMeterMsg) + srcColListSize + exprSize;
return MIN_QUERY_MSG_PKT_SIZE + minMsgSize() + sizeof(SQueryTableMsg) + srcColListSize + exprSize;
}
SMetricMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
SSuperTableMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pMeterMetaInfo->vnodeIndex);
int32_t meterInfoSize = (pMetricMeta->tagLen + sizeof(SMeterSidExtInfo)) * pVnodeSidList->numOfSids;
int32_t meterInfoSize = (pMetricMeta->tagLen + sizeof(STableSidExtInfo)) * pVnodeSidList->numOfSids;
int32_t outputColumnSize = pQueryInfo->fieldsInfo.numOfOutputCols * sizeof(SSqlFuncExprMsg);
int32_t size = meterInfoSize + outputColumnSize + srcColListSize + exprSize + MIN_QUERY_MSG_PKT_SIZE;
@ -1303,34 +1303,34 @@ static int32_t tscEstimateQueryMsgSize(SSqlCmd *pCmd, int32_t clauseIndex) {
return size;
}
static char *doSerializeTableInfo(SSqlObj *pSql, int32_t numOfMeters, int32_t vnodeId, char *pMsg) {
static char *doSerializeTableInfo(SSqlObj *pSql, int32_t numOfTables, int32_t vnodeId, char *pMsg) {
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, pSql->cmd.clauseIndex, 0);
SMeterMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
SMetricMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
STableMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
SSuperTableMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
tscTrace("%p vid:%d, query on %d meters", pSql, htons(vnodeId), numOfMeters);
tscTrace("%p vid:%d, query on %d meters", pSql, htons(vnodeId), numOfTables);
if (UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) {
#ifdef _DEBUG_VIEW
tscTrace("%p sid:%d, uid:%" PRIu64, pSql, pMeterMetaInfo->pMeterMeta->sid, pMeterMetaInfo->pMeterMeta->uid);
#endif
SMeterSidExtInfo *pMeterInfo = (SMeterSidExtInfo *)pMsg;
STableSidExtInfo *pMeterInfo = (STableSidExtInfo *)pMsg;
pMeterInfo->sid = htonl(pMeterMeta->sid);
pMeterInfo->uid = htobe64(pMeterMeta->uid);
pMeterInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pMeterMeta->uid));
pMsg += sizeof(SMeterSidExtInfo);
pMsg += sizeof(STableSidExtInfo);
} else {
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pMeterMetaInfo->vnodeIndex);
for (int32_t i = 0; i < numOfMeters; ++i) {
SMeterSidExtInfo *pMeterInfo = (SMeterSidExtInfo *)pMsg;
SMeterSidExtInfo *pQueryMeterInfo = tscGetMeterSidInfo(pVnodeSidList, i);
for (int32_t i = 0; i < numOfTables; ++i) {
STableSidExtInfo *pMeterInfo = (STableSidExtInfo *)pMsg;
STableSidExtInfo *pQueryMeterInfo = tscGetMeterSidInfo(pVnodeSidList, i);
pMeterInfo->sid = htonl(pQueryMeterInfo->sid);
pMeterInfo->uid = htobe64(pQueryMeterInfo->uid);
pMeterInfo->key = htobe64(tscGetSubscriptionProgress(pSql->pSubscription, pQueryMeterInfo->uid));
pMsg += sizeof(SMeterSidExtInfo);
pMsg += sizeof(STableSidExtInfo);
memcpy(pMsg, pQueryMeterInfo->tags, pMetricMeta->tagLen);
pMsg += pMetricMeta->tagLen;
@ -1359,16 +1359,16 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
char * pStart = pCmd->payload + tsRpcHeadSize;
SMeterMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
SMetricMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
STableMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
SSuperTableMeta *pMetricMeta = pMeterMetaInfo->pMetricMeta;
SQueryMeterMsg *pQueryMsg = (SQueryMeterMsg *)pStart;
SQueryTableMsg *pQueryMsg = (SQueryTableMsg *)pStart;
int32_t msgLen = 0;
int32_t numOfMeters = 0;
int32_t numOfTables = 0;
if (UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) {
numOfMeters = 1;
numOfTables = 1;
tscTrace("%p query on vnode: %d, number of sid:%d, meter id: %s", pSql,
pMeterMeta->vpeerDesc[pMeterMeta->index].vnode, 1, pMeterMetaInfo->name);
@ -1385,17 +1385,17 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, pMeterMetaInfo->vnodeIndex);
uint32_t vnodeId = pVnodeSidList->vpeerDesc[pVnodeSidList->index].vnode;
numOfMeters = pVnodeSidList->numOfSids;
if (numOfMeters <= 0) {
tscError("%p vid:%d,error numOfMeters in query message:%d", pSql, vnodeId, numOfMeters);
numOfTables = pVnodeSidList->numOfSids;
if (numOfTables <= 0) {
tscError("%p vid:%d,error numOfTables in query message:%d", pSql, vnodeId, numOfTables);
return -1; // error
}
tscTrace("%p query on vid:%d, number of sid:%d", pSql, vnodeId, numOfMeters);
tscTrace("%p query on vid:%d, number of sid:%d", pSql, vnodeId, numOfTables);
pQueryMsg->vnode = htons(vnodeId);
}
pQueryMsg->numOfSids = htonl(numOfMeters);
pQueryMsg->numOfSids = htonl(numOfTables);
pQueryMsg->numOfTagsCols = htons(pMeterMetaInfo->numOfTags);
if (pQueryInfo->order.order == TSQL_SO_ASC) {
@ -1566,7 +1566,7 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pQueryMsg->colNameLen = htonl(len);
// serialize the table info (sid, uid, tags)
pMsg = doSerializeTableInfo(pSql, numOfMeters, htons(pQueryMsg->vnode), pMsg);
pMsg = doSerializeTableInfo(pSql, numOfTables, htons(pQueryMsg->vnode), pMsg);
// only include the required tag column schema. If a tag is not required, it won't be sent to vnode
if (pMeterMetaInfo->numOfTags > 0) {
@ -2295,7 +2295,7 @@ int tscProcessTagRetrieveRsp(SSqlObj *pSql) {
int32_t numOfRes = 0;
if (tscSqlExprGet(pQueryInfo, 0)->functionId == TSDB_FUNC_TAGPRJ) {
numOfRes = pMeterMetaInfo->pMetricMeta->numOfMeters;
numOfRes = pMeterMetaInfo->pMetricMeta->numOfTables;
} else {
numOfRes = 1; // for count function, there is only one output.
}
@ -2435,7 +2435,7 @@ int tscBuildMultiMeterMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
assert(pCmd->payloadLen + minMsgSize() <= pCmd->allocSize);
tscTrace("%p build load multi-metermeta msg completed, numOfMeters:%d, msg size:%d", pSql, pCmd->count,
tscTrace("%p build load multi-metermeta msg completed, numOfTables:%d, msg size:%d", pSql, pCmd->count,
pCmd->payloadLen);
return pCmd->payloadLen;
@ -2457,7 +2457,7 @@ static int32_t tscEstimateMetricMetaMsgSize(SSqlCmd *pCmd) {
}
int32_t joinCondLen = (TSDB_TABLE_ID_LEN + sizeof(int16_t)) * 2;
int32_t elemSize = sizeof(SMetricMetaElemMsg) * pQueryInfo->numOfTables;
int32_t elemSize = sizeof(SSuperTableMetaElemMsg) * pQueryInfo->numOfTables;
int32_t len = tagLen + joinCondLen + elemSize + defaultSize;
@ -2492,7 +2492,7 @@ int tscBuildMetricMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
pMsg += sizeof(SMgmtHead);
pMetaMsg = (SSuperTableMetaMsg *)pMsg;
pMetaMsg->numOfMeters = htonl(pQueryInfo->numOfTables);
pMetaMsg->numOfTables = htonl(pQueryInfo->numOfTables);
pMsg += sizeof(SSuperTableMetaMsg);
@ -2521,8 +2521,8 @@ int tscBuildMetricMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
offset = pMsg - (char *)pMetaMsg;
pMetaMsg->metaElem[i] = htonl(offset);
SMetricMetaElemMsg *pElem = (SMetricMetaElemMsg *)pMsg;
pMsg += sizeof(SMetricMetaElemMsg);
SSuperTableMetaElemMsg *pElem = (SSuperTableMetaElemMsg *)pMsg;
pMsg += sizeof(SSuperTableMetaElemMsg);
// convert to unicode before sending to mnode for metric query
int32_t condLen = 0;
@ -2663,7 +2663,7 @@ int tscBuildHeartBeatMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
}
int tscProcessMeterMetaRsp(SSqlObj *pSql) {
SMeterMeta *pMeta;
STableMeta *pMeta;
SSchema * pSchema;
uint8_t ieType;
@ -2676,7 +2676,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) {
}
rsp++;
pMeta = (SMeterMeta *)rsp;
pMeta = (STableMeta *)rsp;
pMeta->sid = htonl(pMeta->sid);
pMeta->sversion = htons(pMeta->sversion);
@ -2705,7 +2705,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) {
}
pMeta->rowSize = 0;
rsp += sizeof(SMeterMeta);
rsp += sizeof(STableMeta);
pSchema = (SSchema *)rsp;
int32_t numOfTotalCols = pMeta->numOfColumns + pMeta->numOfTags;
@ -2741,7 +2741,7 @@ int tscProcessMeterMetaRsp(SSqlObj *pSql) {
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, 0, 0);
assert(pMeterMetaInfo->pMeterMeta == NULL);
pMeterMetaInfo->pMeterMeta = (SMeterMeta *)taosAddDataIntoCache(tscCacheHandle, pMeterMetaInfo->name, (char *)pMeta,
pMeterMetaInfo->pMeterMeta = (STableMeta *)taosAddDataIntoCache(tscCacheHandle, pMeterMetaInfo->name, (char *)pMeta,
size, tsMeterMetaKeepTimer);
// todo handle out of memory case
if (pMeterMetaInfo->pMeterMeta == NULL) return 0;
@ -2777,8 +2777,8 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
rsp += sizeof(SMultiTableInfoMsg);
for (i = 0; i < totalNum; i++) {
SMultiMeterMeta *pMultiMeta = (SMultiMeterMeta *)rsp;
SMeterMeta * pMeta = &pMultiMeta->metas;
SMultiTableMeta *pMultiMeta = (SMultiTableMeta *)rsp;
STableMeta * pMeta = &pMultiMeta->metas;
pMeta->sid = htonl(pMeta->sid);
pMeta->sversion = htons(pMeta->sversion);
@ -2820,7 +2820,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
}
pMeta->rowSize = 0;
rsp += sizeof(SMultiMeterMeta);
rsp += sizeof(SMultiTableMeta);
pSchema = (SSchema *)rsp;
int32_t numOfTotalCols = pMeta->numOfColumns + pMeta->numOfTags;
@ -2847,7 +2847,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
}
rsp += tagLen;
int32_t size = (int32_t)(rsp - ((char *)pMeta)); // Consistent with SMeterMeta in cache
int32_t size = (int32_t)(rsp - ((char *)pMeta)); // Consistent with STableMeta in cache
pMeta->index = 0;
(void)taosAddDataIntoCache(tscCacheHandle, pMeta->tableId, (char *)pMeta, size, tsMeterMetaKeepTimer);
@ -2860,7 +2860,7 @@ int tscProcessMultiMeterMetaRsp(SSqlObj *pSql) {
}
int tscProcessMetricMetaRsp(SSqlObj *pSql) {
SMetricMeta *pMeta;
SSuperTableMeta *pMeta;
uint8_t ieType;
void ** metricMetaList = NULL;
int32_t * sizes = NULL;
@ -2891,16 +2891,16 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
}
for (int32_t k = 0; k < num; ++k) {
pMeta = (SMetricMeta *)rsp;
pMeta = (SSuperTableMeta *)rsp;
size_t size = (size_t)pSql->res.rspLen - 1;
rsp = rsp + sizeof(SMetricMeta);
rsp = rsp + sizeof(SSuperTableMeta);
pMeta->numOfMeters = htonl(pMeta->numOfMeters);
pMeta->numOfTables = htonl(pMeta->numOfTables);
pMeta->numOfVnodes = htonl(pMeta->numOfVnodes);
pMeta->tagLen = htons(pMeta->tagLen);
size += pMeta->numOfVnodes * sizeof(SVnodeSidList *) + pMeta->numOfMeters * sizeof(SMeterSidExtInfo *);
size += pMeta->numOfVnodes * sizeof(SVnodeSidList *) + pMeta->numOfTables * sizeof(STableSidExtInfo *);
char *pBuf = calloc(1, size);
if (pBuf == NULL) {
@ -2908,14 +2908,14 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
goto _error_clean;
}
SMetricMeta *pNewMetricMeta = (SMetricMeta *)pBuf;
SSuperTableMeta *pNewMetricMeta = (SSuperTableMeta *)pBuf;
metricMetaList[k] = pNewMetricMeta;
pNewMetricMeta->numOfMeters = pMeta->numOfMeters;
pNewMetricMeta->numOfTables = pMeta->numOfTables;
pNewMetricMeta->numOfVnodes = pMeta->numOfVnodes;
pNewMetricMeta->tagLen = pMeta->tagLen;
pBuf = pBuf + sizeof(SMetricMeta) + pNewMetricMeta->numOfVnodes * sizeof(SVnodeSidList *);
pBuf = pBuf + sizeof(SSuperTableMeta) + pNewMetricMeta->numOfVnodes * sizeof(SVnodeSidList *);
for (int32_t i = 0; i < pMeta->numOfVnodes; ++i) {
SVnodeSidList *pSidLists = (SVnodeSidList *)rsp;
@ -2924,18 +2924,18 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
pNewMetricMeta->list[i] = pBuf - (char *)pNewMetricMeta; // offset value
SVnodeSidList *pLists = (SVnodeSidList *)pBuf;
tscTrace("%p metricmeta:vid:%d,numOfMeters:%d", pSql, i, pLists->numOfSids);
tscTrace("%p metricmeta:vid:%d,numOfTables:%d", pSql, i, pLists->numOfSids);
pBuf += sizeof(SVnodeSidList) + sizeof(SMeterSidExtInfo *) * pSidLists->numOfSids;
pBuf += sizeof(SVnodeSidList) + sizeof(STableSidExtInfo *) * pSidLists->numOfSids;
rsp += sizeof(SVnodeSidList);
size_t elemSize = sizeof(SMeterSidExtInfo) + pNewMetricMeta->tagLen;
size_t elemSize = sizeof(STableSidExtInfo) + pNewMetricMeta->tagLen;
for (int32_t j = 0; j < pSidLists->numOfSids; ++j) {
pLists->pSidExtInfoList[j] = pBuf - (char *)pLists;
memcpy(pBuf, rsp, elemSize);
((SMeterSidExtInfo *)pBuf)->uid = htobe64(((SMeterSidExtInfo *)pBuf)->uid);
((SMeterSidExtInfo *)pBuf)->sid = htonl(((SMeterSidExtInfo *)pBuf)->sid);
((STableSidExtInfo *)pBuf)->uid = htobe64(((STableSidExtInfo *)pBuf)->uid);
((STableSidExtInfo *)pBuf)->sid = htonl(((STableSidExtInfo *)pBuf)->sid);
rsp += elemSize;
pBuf += elemSize;
@ -2959,7 +2959,7 @@ int tscProcessMetricMetaRsp(SSqlObj *pSql) {
// release the used metricmeta
taosRemoveDataFromCache(tscCacheHandle, (void **)&(pMeterMetaInfo->pMetricMeta), false);
pMeterMetaInfo->pMetricMeta = (SMetricMeta *)taosAddDataIntoCache(tscCacheHandle, name, (char *)metricMetaList[i],
pMeterMetaInfo->pMetricMeta = (SSuperTableMeta *)taosAddDataIntoCache(tscCacheHandle, name, (char *)metricMetaList[i],
sizes[i], tsMetricMetaKeepTimer);
tfree(metricMetaList[i]);
@ -2986,7 +2986,7 @@ _error_clean:
* current process do not use the cache at all
*/
int tscProcessShowRsp(SSqlObj *pSql) {
SMeterMeta * pMeta;
STableMeta * pMeta;
SShowRsp *pShow;
SSchema * pSchema;
char key[20];
@ -3002,11 +3002,11 @@ int tscProcessShowRsp(SSqlObj *pSql) {
pRes->qhandle = pShow->qhandle;
tscResetForNextRetrieve(pRes);
pMeta = &(pShow->meterMeta);
pMeta = &(pShow->tableMeta);
pMeta->numOfColumns = ntohs(pMeta->numOfColumns);
pSchema = (SSchema *)((char *)pMeta + sizeof(SMeterMeta));
pSchema = (SSchema *)((char *)pMeta + sizeof(STableMeta));
pMeta->sid = ntohs(pMeta->sid);
for (int i = 0; i < pMeta->numOfColumns; ++i) {
pSchema->bytes = htons(pSchema->bytes);
@ -3018,9 +3018,9 @@ int tscProcessShowRsp(SSqlObj *pSql) {
taosRemoveDataFromCache(tscCacheHandle, (void *)&(pMeterMetaInfo->pMeterMeta), false);
int32_t size = pMeta->numOfColumns * sizeof(SSchema) + sizeof(SMeterMeta);
int32_t size = pMeta->numOfColumns * sizeof(SSchema) + sizeof(STableMeta);
pMeterMetaInfo->pMeterMeta =
(SMeterMeta *)taosAddDataIntoCache(tscCacheHandle, key, (char *)pMeta, size, tsMeterMetaKeepTimer);
(STableMeta *)taosAddDataIntoCache(tscCacheHandle, key, (char *)pMeta, size, tsMeterMetaKeepTimer);
pCmd->numOfCols = pQueryInfo->fieldsInfo.numOfOutputCols;
SSchema *pMeterSchema = tsGetSchema(pMeterMetaInfo->pMeterMeta);
@ -3078,7 +3078,7 @@ int tscProcessDropDbRsp(SSqlObj *UNUSED_PARAM(pSql)) {
int tscProcessDropTableRsp(SSqlObj *pSql) {
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, 0, 0);
SMeterMeta *pMeterMeta = taosGetDataFromCache(tscCacheHandle, pMeterMetaInfo->name);
STableMeta *pMeterMeta = taosGetDataFromCache(tscCacheHandle, pMeterMetaInfo->name);
if (pMeterMeta == NULL) {
/* not in cache, abort */
return 0;
@ -3105,7 +3105,7 @@ int tscProcessDropTableRsp(SSqlObj *pSql) {
int tscProcessAlterTableMsgRsp(SSqlObj *pSql) {
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, 0, 0);
SMeterMeta *pMeterMeta = taosGetDataFromCache(tscCacheHandle, pMeterMetaInfo->name);
STableMeta *pMeterMeta = taosGetDataFromCache(tscCacheHandle, pMeterMetaInfo->name);
if (pMeterMeta == NULL) { /* not in cache, abort */
return 0;
}
@ -3147,7 +3147,7 @@ int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) {
SSqlCmd *pCmd = &pSql->cmd;
STscObj *pObj = pSql->pTscObj;
SRetrieveMeterRsp *pRetrieve = (SRetrieveMeterRsp *)pRes->pRsp;
SRetrieveTableRsp *pRetrieve = (SRetrieveTableRsp *)pRes->pRsp;
pRes->numOfRows = htonl(pRetrieve->numOfRows);
pRes->precision = htons(pRetrieve->precision);
@ -3166,9 +3166,9 @@ int tscProcessRetrieveRspFromVnode(SSqlObj *pSql) {
char* p = pRes->data + (pField->bytes + offset) * pRes->numOfRows;
int32_t numOfMeters = htonl(*(int32_t*)p);
int32_t numOfTables = htonl(*(int32_t*)p);
p += sizeof(int32_t);
for (int i = 0; i < numOfMeters; i++) {
for (int i = 0; i < numOfTables; i++) {
int64_t uid = htobe64(*(int64_t*)p);
p += sizeof(int64_t);
TSKEY key = htobe64(*(TSKEY*)p);
@ -3189,7 +3189,7 @@ int tscProcessRetrieveRspFromLocal(SSqlObj *pSql) {
SSqlCmd * pCmd = &pSql->cmd;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
SRetrieveMeterRsp *pRetrieve = (SRetrieveMeterRsp *)pRes->pRsp;
SRetrieveTableRsp *pRetrieve = (SRetrieveTableRsp *)pRes->pRsp;
pRes->numOfRows = htonl(pRetrieve->numOfRows);
pRes->data = pRetrieve->data;
@ -3232,7 +3232,7 @@ static int32_t doGetMeterMetaFromServer(SSqlObj *pSql, SMeterMetaInfo *pMeterMet
strcpy(pNewMeterMetaInfo->name, pMeterMetaInfo->name);
memcpy(pNew->cmd.payload, pSql->cmd.payload, TSDB_DEFAULT_PAYLOAD_SIZE); // tag information if table does not exists.
tscTrace("%p new pSqlObj:%p to get meterMeta", pSql, pNew);
tscTrace("%p new pSqlObj:%p to get tableMeta", pSql, pNew);
if (pSql->fp == NULL) {
tsem_init(&pNew->rspSem, 0, 0);
@ -3274,11 +3274,11 @@ int tscGetMeterMeta(SSqlObj *pSql, SMeterMetaInfo *pMeterMetaInfo) {
taosRemoveDataFromCache(tscCacheHandle, (void **)&(pMeterMetaInfo->pMeterMeta), false);
}
pMeterMetaInfo->pMeterMeta = (SMeterMeta *)taosGetDataFromCache(tscCacheHandle, pMeterMetaInfo->name);
pMeterMetaInfo->pMeterMeta = (STableMeta *)taosGetDataFromCache(tscCacheHandle, pMeterMetaInfo->name);
if (pMeterMetaInfo->pMeterMeta != NULL) {
SMeterMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableMeta *pMeterMeta = pMeterMetaInfo->pMeterMeta;
tscTrace("%p retrieve meterMeta from cache, the number of columns:%d, numOfTags:%d", pSql, pMeterMeta->numOfColumns,
tscTrace("%p retrieve tableMeta from cache, the number of columns:%d, numOfTags:%d", pSql, pMeterMeta->numOfColumns,
pMeterMeta->numOfTags);
return TSDB_CODE_SUCCESS;
@ -3374,7 +3374,7 @@ int tscGetMetricMeta(SSqlObj *pSql, int32_t clauseIndex) {
taosRemoveDataFromCache(tscCacheHandle, (void **)&(pMeterMetaInfo->pMetricMeta), false);
SMetricMeta *ppMeta = (SMetricMeta *)taosGetDataFromCache(tscCacheHandle, tagstr);
SSuperTableMeta *ppMeta = (SSuperTableMeta *)taosGetDataFromCache(tscCacheHandle, tagstr);
if (ppMeta == NULL) {
required = true;
break;
@ -3402,7 +3402,7 @@ int tscGetMetricMeta(SSqlObj *pSql, int32_t clauseIndex) {
for (int32_t i = 0; i < pQueryInfo->numOfTables; ++i) {
SMeterMetaInfo *pMMInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, i);
SMeterMeta *pMeterMeta = taosGetDataFromCache(tscCacheHandle, pMMInfo->name);
STableMeta *pMeterMeta = taosGetDataFromCache(tscCacheHandle, pMMInfo->name);
tscAddMeterMetaInfo(pNewQueryInfo, pMMInfo->name, pMeterMeta, NULL, pMMInfo->numOfTags, pMMInfo->tagColumnIndex);
}
@ -3448,7 +3448,7 @@ int tscGetMetricMeta(SSqlObj *pSql, int32_t clauseIndex) {
#endif
taosRemoveDataFromCache(tscCacheHandle, (void **)&(pMeterMetaInfo->pMetricMeta), false);
pMeterMetaInfo->pMetricMeta = (SMetricMeta *)taosGetDataFromCache(tscCacheHandle, tagstr);
pMeterMetaInfo->pMetricMeta = (SSuperTableMeta *)taosGetDataFromCache(tscCacheHandle, tagstr);
}
}

46
src/client/src/tscSub.c
View File

@ -44,7 +44,7 @@ typedef struct SSub {
int interval;
TAOS_SUBSCRIBE_CALLBACK fp;
void * param;
int numOfMeters;
int numOfTables;
SSubscriptionProgress * progress;
} SSub;
@ -62,7 +62,7 @@ TSKEY tscGetSubscriptionProgress(void* sub, int64_t uid) {
return 0;
SSub* pSub = (SSub*)sub;
for (int s = 0, e = pSub->numOfMeters; s < e;) {
for (int s = 0, e = pSub->numOfTables; s < e;) {
int m = (s + e) / 2;
SSubscriptionProgress* p = pSub->progress + m;
if (p->uid > uid)
@ -81,7 +81,7 @@ void tscUpdateSubscriptionProgress(void* sub, int64_t uid, TSKEY ts) {
return;
SSub* pSub = (SSub*)sub;
for (int s = 0, e = pSub->numOfMeters; s < e;) {
for (int s = 0, e = pSub->numOfTables; s < e;) {
int m = (s + e) / 2;
SSubscriptionProgress* p = pSub->progress + m;
if (p->uid > uid)
@ -176,43 +176,43 @@ int tscUpdateSubscription(STscObj* pObj, SSub* pSub) {
}
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0, 0);
int numOfMeters = 0;
int numOfTables = 0;
if (!UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) {
SMetricMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta;
SSuperTableMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta;
for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
numOfMeters += pVnodeSidList->numOfSids;
numOfTables += pVnodeSidList->numOfSids;
}
}
SSubscriptionProgress* progress = (SSubscriptionProgress*)calloc(numOfMeters, sizeof(SSubscriptionProgress));
SSubscriptionProgress* progress = (SSubscriptionProgress*)calloc(numOfTables, sizeof(SSubscriptionProgress));
if (progress == NULL) {
tscError("failed to allocate memory for progress: %s", pSub->topic);
return 0;
}
if (UTIL_METER_IS_NOMRAL_METER(pMeterMetaInfo)) {
numOfMeters = 1;
numOfTables = 1;
int64_t uid = pMeterMetaInfo->pMeterMeta->uid;
progress[0].uid = uid;
progress[0].key = tscGetSubscriptionProgress(pSub, uid);
} else {
SMetricMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta;
numOfMeters = 0;
SSuperTableMeta* pMetricMeta = pMeterMetaInfo->pMetricMeta;
numOfTables = 0;
for (int32_t i = 0; i < pMetricMeta->numOfVnodes; i++) {
SVnodeSidList *pVnodeSidList = tscGetVnodeSidList(pMetricMeta, i);
for (int32_t j = 0; j < pVnodeSidList->numOfSids; j++) {
SMeterSidExtInfo *pMeterInfo = tscGetMeterSidInfo(pVnodeSidList, j);
STableSidExtInfo *pMeterInfo = tscGetMeterSidInfo(pVnodeSidList, j);
int64_t uid = pMeterInfo->uid;
progress[numOfMeters].uid = uid;
progress[numOfMeters++].key = tscGetSubscriptionProgress(pSub, uid);
progress[numOfTables].uid = uid;
progress[numOfTables++].key = tscGetSubscriptionProgress(pSub, uid);
}
}
qsort(progress, numOfMeters, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress);
qsort(progress, numOfTables, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress);
}
free(pSub->progress);
pSub->numOfMeters = numOfMeters;
pSub->numOfTables = numOfTables;
pSub->progress = progress;
pSub->lastSyncTime = taosGetTimestampMs();
@ -257,9 +257,9 @@ static int tscLoadSubscriptionProgress(SSub* pSub) {
return 0;
}
int numOfMeters = atoi(buf);
SSubscriptionProgress* progress = calloc(numOfMeters, sizeof(SSubscriptionProgress));
for (int i = 0; i < numOfMeters; i++) {
int numOfTables = atoi(buf);
SSubscriptionProgress* progress = calloc(numOfTables, sizeof(SSubscriptionProgress));
for (int i = 0; i < numOfTables; i++) {
if (fgets(buf, sizeof(buf), fp) == NULL) {
fclose(fp);
free(progress);
@ -273,10 +273,10 @@ static int tscLoadSubscriptionProgress(SSub* pSub) {
fclose(fp);
qsort(progress, numOfMeters, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress);
pSub->numOfMeters = numOfMeters;
qsort(progress, numOfTables, sizeof(SSubscriptionProgress), tscCompareSubscriptionProgress);
pSub->numOfTables = numOfTables;
pSub->progress = progress;
tscTrace("subscription progress loaded, %d tables: %s", numOfMeters, pSub->topic);
tscTrace("subscription progress loaded, %d tables: %s", numOfTables, pSub->topic);
return 1;
}
@ -297,8 +297,8 @@ void tscSaveSubscriptionProgress(void* sub) {
}
fputs(pSub->pSql->sqlstr, fp);
fprintf(fp, "\n%d\n", pSub->numOfMeters);
for (int i = 0; i < pSub->numOfMeters; i++) {
fprintf(fp, "\n%d\n", pSub->numOfTables);
for (int i = 0; i < pSub->numOfTables; i++) {
int64_t uid = pSub->progress[i].uid;
TSKEY key = pSub->progress[i].key;
fprintf(fp, "%" PRId64 ":%" PRId64 "\n", uid, key);

28
src/client/src/tscUtil.c
View File

@ -170,13 +170,13 @@ void tscGetDBInfoFromMeterId(char* tableId, char* db) {
db[0] = 0;
}
SVnodeSidList* tscGetVnodeSidList(SMetricMeta* pMetricmeta, int32_t vnodeIdx) {
SVnodeSidList* tscGetVnodeSidList(SSuperTableMeta* pMetricmeta, int32_t vnodeIdx) {
if (pMetricmeta == NULL) {
tscError("illegal metricmeta");
return 0;
}
if (pMetricmeta->numOfVnodes == 0 || pMetricmeta->numOfMeters == 0) {
if (pMetricmeta->numOfVnodes == 0 || pMetricmeta->numOfTables == 0) {
return 0;
}
@ -190,7 +190,7 @@ SVnodeSidList* tscGetVnodeSidList(SMetricMeta* pMetricmeta, int32_t vnodeIdx) {
return (SVnodeSidList*)(pMetricmeta->list[vnodeIdx] + (char*)pMetricmeta);
}
SMeterSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
STableSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
if (pSidList == NULL) {
tscError("illegal sidlist");
return 0;
@ -202,7 +202,7 @@ SMeterSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx) {
tscError("illegal sidIdx:%d, reset to 0, sidIdx range:%d-%d", idx, 0, sidRange);
idx = 0;
}
return (SMeterSidExtInfo*)(pSidList->pSidExtInfoList[idx] + (char*)pSidList);
return (STableSidExtInfo*)(pSidList->pSidExtInfoList[idx] + (char*)pSidList);
}
bool tscIsTwoStageMergeMetricQuery(SQueryInfo* pQueryInfo, int32_t tableIndex) {
@ -583,7 +583,7 @@ int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock) {
SSqlCmd* pCmd = &pSql->cmd;
assert(pDataBlock->pMeterMeta != NULL);
pCmd->numOfTablesInSubmit = pDataBlock->numOfMeters;
pCmd->numOfTablesInSubmit = pDataBlock->numOfTables;
assert(pCmd->numOfClause == 1);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, pCmd->clauseIndex, 0);
@ -639,7 +639,7 @@ void tscFreeUnusedDataBlocks(SDataBlockList* pList) {
* @return
*/
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
SMeterMeta* pMeterMeta, STableDataBlocks** dataBlocks) {
STableMeta* pMeterMeta, STableDataBlocks** dataBlocks) {
STableDataBlocks* dataBuf = (STableDataBlocks*)calloc(1, sizeof(STableDataBlocks));
if (dataBuf == NULL) {
tscError("failed to allocated memory, reason:%s", strerror(errno));
@ -675,7 +675,7 @@ int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOff
}
int32_t tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList, int64_t id, int32_t size,
int32_t startOffset, int32_t rowSize, const char* tableId, SMeterMeta* pMeterMeta,
int32_t startOffset, int32_t rowSize, const char* tableId, STableMeta* pMeterMeta,
STableDataBlocks** dataBlocks) {
*dataBlocks = NULL;
@ -754,7 +754,7 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SDataBlockList* pTableDataBlockLi
memcpy(dataBuf->pData + dataBuf->size, pOneTableBlock->pData, pOneTableBlock->size);
dataBuf->size += pOneTableBlock->size;
dataBuf->numOfMeters += 1;
dataBuf->numOfTables += 1;
}
tscDestroyBlockArrayList(pTableDataBlockList);
@ -1799,8 +1799,8 @@ void tscFreeSubqueryInfo(SSqlCmd* pCmd) {
tfree(pCmd->pQueryInfo);
}
SMeterMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, SMeterMeta* pMeterMeta,
SMetricMeta* pMetricMeta, int16_t numOfTags, int16_t* tags) {
SMeterMetaInfo* tscAddMeterMetaInfo(SQueryInfo* pQueryInfo, const char* name, STableMeta* pMeterMeta,
SSuperTableMeta* pMetricMeta, int16_t numOfTags, int16_t* tags) {
void* pAlloc = realloc(pQueryInfo->pMeterInfo, (pQueryInfo->numOfTables + 1) * POINTER_BYTES);
if (pAlloc == NULL) {
return NULL;
@ -1986,16 +1986,16 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
SMeterMetaInfo* pFinalInfo = NULL;
if (pPrevSql == NULL) {
SMeterMeta* pMeterMeta = taosGetDataFromCache(tscCacheHandle, name);
SMetricMeta* pMetricMeta = taosGetDataFromCache(tscCacheHandle, key);
STableMeta* pMeterMeta = taosGetDataFromCache(tscCacheHandle, name);
SSuperTableMeta* pMetricMeta = taosGetDataFromCache(tscCacheHandle, key);
pFinalInfo = tscAddMeterMetaInfo(pNewQueryInfo, name, pMeterMeta, pMetricMeta, pMeterMetaInfo->numOfTags,
pMeterMetaInfo->tagColumnIndex);
} else { // transfer the ownership of pMeterMeta/pMetricMeta to the newly create sql object.
SMeterMetaInfo* pPrevInfo = tscGetMeterMetaInfo(&pPrevSql->cmd, pPrevSql->cmd.clauseIndex, 0);
SMeterMeta* pPrevMeterMeta = taosTransferDataInCache(tscCacheHandle, (void**)&pPrevInfo->pMeterMeta);
SMetricMeta* pPrevMetricMeta = taosTransferDataInCache(tscCacheHandle, (void**)&pPrevInfo->pMetricMeta);
STableMeta* pPrevMeterMeta = taosTransferDataInCache(tscCacheHandle, (void**)&pPrevInfo->pMeterMeta);
SSuperTableMeta* pPrevMetricMeta = taosTransferDataInCache(tscCacheHandle, (void**)&pPrevInfo->pMetricMeta);
pFinalInfo = tscAddMeterMetaInfo(pNewQueryInfo, name, pPrevMeterMeta, pPrevMetricMeta, pMeterMetaInfo->numOfTags,
pMeterMetaInfo->tagColumnIndex);

6
src/dnode/inc/dnodeRead.h
View File

@ -38,18 +38,18 @@ void dnodeFreeQInfos();
/*
* handle query message, and the result is returned by callback function
*/
void dnodeQueryData(SQueryMeterMsg *pQuery, void *pConn, void (*callback)(int32_t code, void *pQInfo, void *pConn));
void dnodeQueryData(SQueryTableMsg *pQuery, void *pConn, void (*callback)(int32_t code, void *pQInfo, void *pConn));
/*
* Dispose retrieve msg, and the result will passed through callback function
*/
typedef void (*SDnodeRetrieveCallbackFp)(int32_t code, void *pQInfo, void *pConn);
void dnodeRetrieveData(SRetrieveMeterMsg *pRetrieve, void *pConn, SDnodeRetrieveCallbackFp callbackFp);
void dnodeRetrieveData(SRetrieveTableMsg *pRetrieve, void *pConn, SDnodeRetrieveCallbackFp callbackFp);
/*
* Fill retrieve result according to query info
*/
int32_t dnodeGetRetrieveData(void *pQInfo, SRetrieveMeterRsp *retrievalRsp);
int32_t dnodeGetRetrieveData(void *pQInfo, SRetrieveTableRsp *retrievalRsp);
/*
* Get the size of retrieve result according to query info

4
src/dnode/src/dnodeMgmt.c
View File

@ -193,13 +193,13 @@ void dnodeSendVpeerCfgMsg(int32_t vnode) {
}
void dnodeSendMeterCfgMsg(int32_t vnode, int32_t sid) {
SMeterCfgMsg *cfg = (SMeterCfgMsg *) rpcMallocCont(sizeof(SMeterCfgMsg));
STableCfgMsg *cfg = (STableCfgMsg *) rpcMallocCont(sizeof(STableCfgMsg));
if (cfg == NULL) {
return;
}
cfg->vnode = htonl(vnode);
dnodeSendMsgToMnode((int8_t*)cfg, sizeof(SMeterCfgMsg), TSDB_MSG_TYPE_TABLE_CFG);
dnodeSendMsgToMnode((int8_t*)cfg, sizeof(STableCfgMsg), TSDB_MSG_TYPE_TABLE_CFG);
}
void dnodeInitProcessShellMsg() {

12
src/dnode/src/dnodeRead.c
View File

@ -26,14 +26,14 @@ void dnodeFreeQInfo(void *pConn) {}
void dnodeFreeQInfos() {}
void dnodeQueryData(SQueryMeterMsg *pQuery, void *pConn, void (*callback)(int32_t code, void *pQInfo, void *pConn)) {
void dnodeQueryData(SQueryTableMsg *pQuery, void *pConn, void (*callback)(int32_t code, void *pQInfo, void *pConn)) {
void *pQInfo = NULL;
int code = TSDB_CODE_SUCCESS;
callback(code, pConn, pQInfo);
}
static void dnodeExecuteRetrieveData(SSchedMsg *pSched) {
//SRetrieveMeterMsg *pRetrieve = (SRetrieveMeterMsg *)pSched->msg;
//SRetrieveTableMsg *pRetrieve = (SRetrieveTableMsg *)pSched->msg;
SDnodeRetrieveCallbackFp callback = (SDnodeRetrieveCallbackFp)pSched->thandle;
void *pConn = pSched->ahandle;
@ -47,9 +47,9 @@ static void dnodeExecuteRetrieveData(SSchedMsg *pSched) {
free(pSched->msg);
}
void dnodeRetrieveData(SRetrieveMeterMsg *pRetrieve, void *pConn, SDnodeRetrieveCallbackFp callbackFp) {
int8_t *msg = malloc(sizeof(SRetrieveMeterMsg));
memcpy(msg, pRetrieve, sizeof(SRetrieveMeterMsg));
void dnodeRetrieveData(SRetrieveTableMsg *pRetrieve, void *pConn, SDnodeRetrieveCallbackFp callbackFp) {
int8_t *msg = malloc(sizeof(SRetrieveTableMsg));
memcpy(msg, pRetrieve, sizeof(SRetrieveTableMsg));
SSchedMsg schedMsg;
schedMsg.msg = msg;
@ -59,7 +59,7 @@ void dnodeRetrieveData(SRetrieveMeterMsg *pRetrieve, void *pConn, SDnodeRetrieve
taosScheduleTask(tsQueryQhandle, &schedMsg);
}
int32_t dnodeGetRetrieveData(void *pQInfo, SRetrieveMeterRsp *retrievalRsp) {
int32_t dnodeGetRetrieveData(void *pQInfo, SRetrieveTableRsp *retrievalRsp) {
return 0;
}

10
src/dnode/src/dnodeShell.c
View File

@ -107,8 +107,8 @@ void dnodeCleanupShell() {
}
void dnodeProcessQueryRequestCb(int code, void *pQInfo, void *pConn) {
int32_t contLen = sizeof(SQueryMeterRsp);
SQueryMeterRsp *queryRsp = (SQueryMeterRsp *) rpcMallocCont(contLen);
int32_t contLen = sizeof(SQueryTableRsp);
SQueryTableRsp *queryRsp = (SQueryTableRsp *) rpcMallocCont(contLen);
if (queryRsp == NULL) {
return;
}
@ -125,7 +125,7 @@ static void dnodeProcessQueryRequest(int8_t *pCont, int32_t contLen, void *pConn
atomic_fetch_add_32(&tsDnodeQueryReqNum, 1);
dTrace("conn:%p, start to query data", pConn);
SQueryMeterMsg *pQuery = (SQueryMeterMsg *) pCont;
SQueryTableMsg *pQuery = (SQueryTableMsg *) pCont;
dnodeQueryData(pQuery, pConn, dnodeProcessQueryRequestCb);
}
@ -140,7 +140,7 @@ void dnodeProcessRetrieveRequestCb(int32_t code, void *pQInfo, void *pConn) {
assert(pQInfo != NULL);
int32_t contLen = dnodeGetRetrieveDataSize(pQInfo);
SRetrieveMeterRsp *retrieveRsp = (SRetrieveMeterRsp *) rpcMallocCont(contLen);
SRetrieveTableRsp *retrieveRsp = (SRetrieveTableRsp *) rpcMallocCont(contLen);
if (retrieveRsp == NULL) {
rpcSendResponse(pConn, TSDB_CODE_SERV_OUT_OF_MEMORY, 0, 0);
return;
@ -162,7 +162,7 @@ void dnodeProcessRetrieveRequestCb(int32_t code, void *pQInfo, void *pConn) {
static void dnodeProcessRetrieveRequest(int8_t *pCont, int32_t contLen, void *pConn) {
dTrace("conn:%p, start to retrieve data", pConn);
SRetrieveMeterMsg *pRetrieve = (SRetrieveMeterMsg *) pCont;
SRetrieveTableMsg *pRetrieve = (SRetrieveTableMsg *) pCont;
dnodeRetrieveData(pRetrieve, pConn, dnodeProcessRetrieveRequestCb);
}

4
src/inc/mnode.h
View File

@ -102,14 +102,14 @@ typedef struct {
} STableGid;
typedef struct _tab_obj {
char meterId[TSDB_TABLE_ID_LEN + 1];
char tableId[TSDB_TABLE_ID_LEN + 1];
uint64_t uid;
STableGid gid;
int32_t sversion; // schema version
int64_t createdTime;
int32_t numOfTags; // for metric
int32_t numOfMeters; // for metric
int32_t numOfTables; // for metric
int32_t numOfColumns;
int32_t schemaSize;
short nextColId;

80
src/inc/taosmsg.h
View File

@ -190,29 +190,6 @@ extern char *taosMsg[];
#pragma pack(push, 1)
typedef struct {
uint32_t customerId;
uint32_t osId;
uint32_t appId;
char hwId[TSDB_UNI_LEN];
char hwVersion[TSDB_VERSION_LEN];
char osVersion[TSDB_VERSION_LEN];
char appVersion[TSDB_VERSION_LEN];
char sdkVersion[TSDB_VERSION_LEN];
char name[TSDB_UNI_LEN];
char street[TSDB_STREET_LEN];
char city[TSDB_CITY_LEN];
char state[TSDB_STATE_LEN];
char country[TSDB_COUNTRY_LEN];
uint32_t longitude;
uint32_t latitude;
} SRegMsg;
typedef struct {
short numOfRows;
char payLoad[];
} SSubmitMsg;
typedef struct {
int32_t vnode;
int32_t sid;
@ -226,7 +203,7 @@ typedef struct {
int8_t import;
int8_t reserved[3];
int32_t numOfSid; /* total number of sid */
char blks[]; /* numOfSid blocks, each blocks for one meter */
char blks[]; /* numOfSid blocks, each blocks for one table */
} SShellSubmitMsg;
typedef struct {
@ -360,7 +337,7 @@ typedef struct SColIndexEx {
/*
* colIdx is the index of column in latest schema of table
* it is available in the client side. Also used to determine
* whether current meter schema is up-to-date.
* whether current table schema is up-to-date.
*
* colIdxInBuf is used to denote the index of column in pQuery->colList,
* this value is invalid in client side, as well as in cache block of vnode either.
@ -438,22 +415,22 @@ typedef struct SColumnInfo {
/*
* enable vnode to understand how to group several tables with different tag;
*/
typedef struct SMeterSidExtInfo {
typedef struct STableSidExtInfo {
int32_t sid;
int64_t uid;
TSKEY key; // key for subscription
char tags[];
} SMeterSidExtInfo;
} STableSidExtInfo;
/*
* the outputCols is equalled to or larger than numOfCols
* e.g., select min(colName), max(colName), avg(colName) from meter_name
* e.g., select min(colName), max(colName), avg(colName) from table
* the outputCols will be 3 while the numOfCols is 1.
*/
typedef struct {
int16_t vnode;
int32_t numOfSids;
uint64_t pSidExtInfo; // meter id & tag info ptr, in windows pointer may
uint64_t pSidExtInfo; // table id & tag info ptr, in windows pointer may
uint64_t uid;
TSKEY skey;
@ -498,26 +475,17 @@ typedef struct {
int32_t tsNumOfBlocks; // ts comp block numbers
int32_t tsOrder; // ts comp block order
SColumnInfo colList[];
} SQueryMeterMsg;
} SQueryTableMsg;
typedef struct {
char code;
uint64_t qhandle;
} SQueryMeterRsp;
typedef struct {
TSKEY skey;
TSKEY ekey;
int32_t num;
short order;
short numOfCols;
short colList[];
} SQueryMsg;
} SQueryTableRsp;
typedef struct {
uint64_t qhandle;
uint16_t free;
} SRetrieveMeterMsg;
} SRetrieveTableMsg;
typedef struct {
int32_t numOfRows;
@ -525,7 +493,7 @@ typedef struct {
int64_t offset; // updated offset value for multi-vnode projection query
int64_t useconds;
char data[];
} SRetrieveMeterRsp;
} SRetrieveTableRsp;
typedef struct {
uint32_t vnode;
@ -564,7 +532,7 @@ typedef struct {
int32_t daysToKeep;
int32_t commitTime;
int32_t rowsInFileBlock;
int16_t blocksPerMeter;
int16_t blocksPerTable;
int8_t compression;
int8_t commitLog;
int8_t replications;
@ -678,10 +646,10 @@ typedef struct {
int16_t numOfGroupCols; // num of group by columns
int32_t groupbyTagColumnList;
} SMetricMetaElemMsg;
} SSuperTableMetaElemMsg;
typedef struct {
int32_t numOfMeters;
int32_t numOfTables;
int32_t join;
int32_t joinCondLen; // for join condition
int32_t metaElem[TSDB_MAX_JOIN_TABLE_NUM];
@ -691,17 +659,17 @@ typedef struct {
SVPeerDesc vpeerDesc[TSDB_VNODES_SUPPORT];
int16_t index; // used locally
int32_t numOfSids;
int32_t pSidExtInfoList[]; // offset value of SMeterSidExtInfo
int32_t pSidExtInfoList[]; // offset value of STableSidExtInfo
} SVnodeSidList;
typedef struct {
int32_t numOfMeters;
int32_t numOfTables;
int32_t numOfVnodes;
uint16_t tagLen; /* tag value length */
int32_t list[]; /* offset of SVnodeSidList, compared to the SMetricMeta struct */
} SMetricMeta;
int32_t list[]; /* offset of SVnodeSidList, compared to the SSuperTableMeta struct */
} SSuperTableMeta;
typedef struct SMeterMeta {
typedef struct STableMeta {
char tableId[TSDB_TABLE_ID_LEN]; // note: This field must be at the front
int32_t contLen;
uint8_t numOfTags : 6;
@ -716,13 +684,13 @@ typedef struct SMeterMeta {
int32_t vgid;
uint64_t uid;
SSchema schema[];
} SMeterMeta;
} STableMeta;
typedef struct SMultiMeterMeta {
typedef struct SMultiTableMeta {
int32_t numOfTables;
int32_t contLen;
SMeterMeta metas[];
} SMultiMeterMeta;
STableMeta metas[];
} SMultiTableMeta;
typedef struct {
char name[TSDB_TABLE_ID_LEN];
@ -743,7 +711,7 @@ typedef struct {
typedef struct {
uint64_t qhandle;
SMeterMeta meterMeta;
STableMeta tableMeta;
} SShowRsp;
typedef struct {
@ -753,7 +721,7 @@ typedef struct {
typedef struct {
int32_t vnode;
int32_t sid;
} SMeterCfgMsg;
} STableCfgMsg;
typedef struct {
int32_t vnode;

2
src/inc/trpc.h
View File

@ -65,7 +65,7 @@ typedef struct {
void (*ufp)(void *ahandle, SRpcIpSet *pIpSet);
// call back to retrieve the client auth info, for server app only
int (*afp)(char *meterId, char *spi, char *encrypt, char *secret, char *ckey);
int (*afp)(char *tableId, char *spi, char *encrypt, char *secret, char *ckey);
} SRpcInit;
void *rpcOpen(SRpcInit *pRpc);

2
src/mnode/inc/mgmtAcct.h
View File

@ -33,7 +33,7 @@ extern SAcctObj* (*mgmtGetAcct)(char *acctName);
extern int32_t (*mgmtCheckUserLimit)(SAcctObj *pAcct);
extern int32_t (*mgmtCheckDbLimit)(SAcctObj *pAcct);
extern int32_t (*mgmtCheckTableLimit)(SAcctObj *pAcct, SCreateTableMsg *pCreate);
extern int32_t (*mgmtGetAcctMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtGetAcctMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveAccts)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
#ifdef __cplusplus

2
src/mnode/inc/mgmtChildTable.h
View File

@ -37,7 +37,7 @@ int32_t mgmtAlterChildTable(SDbObj *pDb, SAlterTableMsg *pAlter);
int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName, char *nContent);
int8_t *mgmtBuildCreateChildTableMsg(SChildTableObj *pTable, SVgObj *pVgroup);
int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp);
int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMeta *pMeta, bool usePublicIp);
#ifdef __cplusplus
}

2
src/mnode/inc/mgmtConn.h
View File

@ -22,7 +22,7 @@ extern "C" {
#include "mnode.h"
int mgmtGetConnsMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int mgmtGetConnsMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int mgmtRetrieveConns(SShowObj *pShow, char *data, int rows, void *pConn);
bool mgmtCheckQhandle(uint64_t qhandle);

2
src/mnode/inc/mgmtDb.h
View File

@ -29,7 +29,7 @@ int32_t mgmtAddVgroupIntoDbTail(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtRemoveVgroupFromDb(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtMoveVgroupToTail(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtMoveVgroupToHead(SDbObj *pDb, SVgObj *pVgroup);
int32_t mgmtGetDbMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetDbMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
void mgmtCleanUpDbs();

10
src/mnode/inc/mgmtDnode.h
View File

@ -30,18 +30,18 @@ int32_t mgmtDropDnodeByIp(uint32_t ip);
int32_t mgmtGetNextVnode(SVnodeGid *pVnodeGid);
void mgmtSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes, int32_t vgId);
void mgmtUnSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes);
int32_t mgmtGetDnodeMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetDnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
int32_t mgmtSendCfgDnodeMsg(char *cont);
void mgmtSetDnodeMaxVnodes(SDnodeObj *pDnode);
int32_t mgmtGetConfigMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetConfigMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, void *pConn);
int32_t mgmtGetModuleMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetModuleMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pConn);
int32_t mgmtGetVnodeMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveVnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);
extern int32_t (*mgmtInitDnodes)();
@ -51,7 +51,7 @@ extern int32_t (*mgmtGetDnodesNum)();
extern void* (*mgmtGetNextDnode)(SShowObj *pShow, SDnodeObj **pDnode);
extern int32_t (*mgmtUpdateDnode)(SDnodeObj *pDnode);
extern void (*mgmtSetDnodeUnRemove)(SDnodeObj *pDnode);
extern int32_t (*mgmtGetScoresMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtGetScoresMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveScores)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
extern bool (*mgmtCheckConfigShow)(SGlobalConfig *cfg);

2
src/mnode/inc/mgmtGrant.h
View File

@ -30,7 +30,7 @@ extern void (*mgmtRestoreTimeSeries)(uint32_t timeseries);
extern int32_t (*mgmtCheckTimeSeries)(uint32_t timeseries);
extern int32_t (*mgmtCheckUserGrant)();
extern int32_t (*mgmtCheckDbGrant)();
extern int32_t (*mgmtGetGrantsMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtGetGrantsMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveGrants)(SShowObj *pShow, char *data, int rows, void *pConn);
#ifdef __cplusplus

2
src/mnode/inc/mgmtMnode.h
View File

@ -24,7 +24,7 @@ extern "C" {
#include <stdbool.h>
#include "mnode.h"
extern int32_t (*mgmtGetMnodeMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtGetMnodeMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
extern int32_t (*mgmtRetrieveMnodes)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
#ifdef __cplusplus

2
src/mnode/inc/mgmtNormalTable.h
View File

@ -35,7 +35,7 @@ int32_t mgmtAddNormalTableColumn(SNormalTableObj *pTable, SSchema schema[], int3
int32_t mgmtDropNormalTableColumnByName(SNormalTableObj *pTable, char *colName);
int8_t *mgmtBuildCreateNormalTableMsg(SNormalTableObj *pTable);
int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp);
int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMeta *pMeta, bool usePublicIp);
#ifdef __cplusplus
}

4
src/mnode/inc/mgmtProfile.h
View File

@ -22,9 +22,9 @@ extern "C" {
#include "mnode.h"
int32_t mgmtGetQueryMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetQueryMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetStreamMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetStreamMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveQueries(SShowObj *pShow, char *data, int32_t rows, void *pConn);

2
src/mnode/inc/mgmtStreamTable.h
View File

@ -34,7 +34,7 @@ int32_t mgmtDropStreamTable(SDbObj *pDb, SStreamTableObj *pTable);
int32_t mgmtAlterStreamTable(SDbObj *pDb, SAlterTableMsg *pAlter);
int8_t *mgmtBuildCreateStreamTableMsg(SStreamTableObj *pTable, SVgObj *pVgroup);
int32_t mgmtGetStreamTableMeta(SDbObj *pDb, SStreamTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp);
int32_t mgmtGetStreamTableMeta(SDbObj *pDb, SStreamTableObj *pTable, STableMeta *pMeta, bool usePublicIp);
#ifdef __cplusplus
}

4
src/mnode/inc/mgmtSuperTable.h
View File

@ -30,7 +30,7 @@ int32_t mgmtInitSuperTables();
void mgmtCleanUpSuperTables();
void * mgmtGetSuperTable(char *tableId);
int32_t mgmtGetShowSuperTableMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetShowSuperTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveShowSuperTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
int32_t mgmtCreateSuperTable(SDbObj *pDb, SCreateTableMsg *pCreate);
@ -41,7 +41,7 @@ int32_t mgmtModifySuperTableTagNameByName(SSuperTableObj *pTable, char *oldTagNa
int32_t mgmtAddSuperTableColumn(SSuperTableObj *pTable, SSchema schema[], int32_t ncols);
int32_t mgmtDropSuperTableColumnByName(SSuperTableObj *pTable, char *colName);
int32_t mgmtGetSuperTableMeta(SDbObj *pDb, SSuperTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp);
int32_t mgmtGetSuperTableMeta(SDbObj *pDb, SSuperTableObj *pTable, STableMeta *pMeta, bool usePublicIp);
int32_t mgmtFindSuperTableTagIndex(SSuperTableObj *pTable, const char *tagName);
int32_t mgmtSetSchemaFromSuperTable(SSchema *pSchema, SSuperTableObj *pTable);

6
src/mnode/inc/mgmtTable.h
View File

@ -28,13 +28,13 @@ extern "C" {
int32_t mgmtInitTables();
STableInfo* mgmtGetTable(char *tableId);
STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid);
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, SMeterMeta *pMeta, bool usePublicIp);
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp);
int32_t mgmtRetrieveMetricMeta(void *pConn, char **pStart, SSuperTableMetaMsg *pInfo);
int32_t mgmtCreateTable(SDbObj *pDb, SCreateTableMsg *pCreate);
int32_t mgmtDropTable(SDbObj *pDb, char *meterId, int32_t ignore);
int32_t mgmtDropTable(SDbObj *pDb, char *tableId, int32_t ignore);
int32_t mgmtAlterTable(SDbObj *pDb, SAlterTableMsg *pAlter);
int32_t mgmtGetShowTableMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveShowTables(SShowObj *pShow, char *data, int32_t rows, void *pConn);
void mgmtCleanUpMeters();

2
src/mnode/inc/mgmtUser.h
View File

@ -29,7 +29,7 @@ SUserObj *mgmtGetUser(char *name);
int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass);
int32_t mgmtDropUser(SAcctObj *pAcct, char *name);
int32_t mgmtUpdateUser(SUserObj *pUser);
int32_t mgmtGetUserMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetUserMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveUsers(SShowObj *pShow, char *data, int32_t rows, void *pConn);
void mgmtCleanUpUsers();
SUserObj *mgmtGetUserFromConn(void *pConn);

2
src/mnode/inc/mgmtVgroup.h
View File

@ -29,7 +29,7 @@ SVgObj *mgmtGetVgroup(int32_t vgId);
SVgObj *mgmtCreateVgroup(SDbObj *pDb);
int32_t mgmtDropVgroup(SDbObj *pDb, SVgObj *pVgroup);
void mgmtSetVgroupIdPool();
int32_t mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn);
int32_t mgmtRetrieveVgroups(SShowObj *pShow, char *data, int32_t rows, void *pConn);
void mgmtCleanUpVgroups();

4
src/mnode/src/mgmtAcct.c
View File

@ -144,11 +144,11 @@ static void mgmtCleanUpAcctsImp() {
void (*mgmtCleanUpAccts)() = mgmtCleanUpAcctsImp;
static int32_t mgmtGetAcctMetaImp(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
static int32_t mgmtGetAcctMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
int32_t (*mgmtGetAcctMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetAcctMetaImp;
int32_t (*mgmtGetAcctMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetAcctMetaImp;
static int32_t mgmtRetrieveAcctsImp(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
return 0;

8
src/mnode/src/mgmtChildTable.c
View File

@ -383,16 +383,16 @@ int32_t mgmtModifyChildTableTagValueByName(SChildTableObj *pTable, char *tagName
// if (pTable->isDirty) pTable->isDirty = 0;
//
// if (ret < 0) {
// mError("Failed to modify tag column %d of table %s", col, pTable->meterId);
// mError("Failed to modify tag column %d of table %s", col, pTable->tableId);
// return TSDB_CODE_APP_ERROR;
// }
//
// mTrace("Succeed to modify tag column %d of table %s", col, pTable->meterId);
// mTrace("Succeed to modify tag column %d of table %s", col, pTable->tableId);
// return TSDB_CODE_SUCCESS;
return 0;
}
int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp) {
int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, STableMeta *pMeta, bool usePublicIp) {
pMeta->uid = htobe64(pTable->uid);
pMeta->sid = htonl(pTable->sid);
pMeta->vgid = htonl(pTable->vgId);
@ -401,7 +401,7 @@ int32_t mgmtGetChildTableMeta(SDbObj *pDb, SChildTableObj *pTable, SMeterMeta *p
pMeta->numOfTags = pTable->superTable->numOfTags;
pMeta->numOfColumns = htons(pTable->superTable->numOfColumns);
pMeta->tableType = pTable->type;
pMeta->contLen = sizeof(SMeterMeta) + mgmtSetSchemaFromSuperTable(pMeta->schema, pTable->superTable);
pMeta->contLen = sizeof(STableMeta) + mgmtSetSchemaFromSuperTable(pMeta->schema, pTable->superTable);
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {

2
src/mnode/src/mgmtConn.c
View File

@ -67,7 +67,7 @@ int mgmtGetConns(SShowObj *pShow, void *pConn) {
return 0;
}
int mgmtGetConnsMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int mgmtGetConnsMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int cols = 0;
pShow->bytes[cols] = TSDB_METER_NAME_LEN;

24
src/mnode/src/mgmtDb.c
View File

@ -105,13 +105,13 @@ SDbObj *mgmtGetDb(char *db) {
return (SDbObj *)sdbGetRow(tsDbSdb, db);
}
SDbObj *mgmtGetDbByTableId(char *meterId) {
SDbObj *mgmtGetDbByTableId(char *tableId) {
char db[TSDB_TABLE_ID_LEN], *pos;
pos = strstr(meterId, TS_PATH_DELIMITER);
pos = strstr(tableId, TS_PATH_DELIMITER);
pos = strstr(pos + 1, TS_PATH_DELIMITER);
memset(db, 0, sizeof(db));
strncpy(db, meterId, pos - meterId);
strncpy(db, tableId, pos - tableId);
return (SDbObj *)sdbGetRow(tsDbSdb, db);
}
@ -217,16 +217,16 @@ int32_t mgmtCheckDbParams(SCreateDbMsg *pCreate) {
}
// calculate the blocks per table
if (pCreate->blocksPerMeter < 0) {
pCreate->blocksPerMeter = pCreate->cacheNumOfBlocks.totalBlocks / 4;
if (pCreate->blocksPerTable < 0) {
pCreate->blocksPerTable = pCreate->cacheNumOfBlocks.totalBlocks / 4;
}
if (pCreate->blocksPerMeter > pCreate->cacheNumOfBlocks.totalBlocks * 3 / 4) {
pCreate->blocksPerMeter = pCreate->cacheNumOfBlocks.totalBlocks * 3 / 4;
if (pCreate->blocksPerTable > pCreate->cacheNumOfBlocks.totalBlocks * 3 / 4) {
pCreate->blocksPerTable = pCreate->cacheNumOfBlocks.totalBlocks * 3 / 4;
}
if (pCreate->blocksPerMeter < TSDB_MIN_AVG_BLOCKS) {
pCreate->blocksPerMeter = TSDB_MIN_AVG_BLOCKS;
if (pCreate->blocksPerTable < TSDB_MIN_AVG_BLOCKS) {
pCreate->blocksPerTable = TSDB_MIN_AVG_BLOCKS;
}
pCreate->maxSessions++;
@ -340,7 +340,7 @@ void mgmtDropDbFromSdb(SDbObj *pDb) {
// SSuperTableObj *pMetric = pDb->pSTable;
// while (pMetric) {
// SSuperTableObj *pNext = pMetric->next;
// mgmtDropTable(pDb, pMetric->meterId, 0);
// mgmtDropTable(pDb, pMetric->tableId, 0);
// pMetric = pNext;
// }
@ -527,7 +527,7 @@ void mgmtCleanUpDbs() {
sdbCloseTable(tsDbSdb);
}
int32_t mgmtGetDbMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetDbMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SSchema *pSchema = tsGetSchema(pMeta);
@ -760,7 +760,7 @@ int32_t mgmtRetrieveDbs(SShowObj *pShow, char *data, int32_t rows, void *pConn)
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
*(int16_t *)pWrite = pDb->cfg.blocksPerMeter;
*(int16_t *)pWrite = pDb->cfg.blocksPerTable;
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;

12
src/mnode/src/mgmtDnode.c
View File

@ -97,7 +97,7 @@ void mgmtUnSetDnodeVgid(SVnodeGid vnodeGid[], int32_t numOfVnodes) {
}
}
int32_t mgmtGetDnodeMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetDnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SUserObj *pUser = mgmtGetUserFromConn(pConn);
@ -212,7 +212,7 @@ int32_t mgmtRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, void *pCon
return numOfRows;
}
int32_t mgmtGetModuleMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetModuleMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SUserObj *pUser = mgmtGetUserFromConn(pConn);
@ -305,7 +305,7 @@ int32_t mgmtRetrieveModules(SShowObj *pShow, char *data, int32_t rows, void *pCo
return numOfRows;
}
int32_t mgmtGetConfigMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetConfigMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SUserObj *pUser = mgmtGetUserFromConn(pConn);
@ -393,7 +393,7 @@ int32_t mgmtRetrieveConfigs(SShowObj *pShow, char *data, int32_t rows, void *pCo
return numOfRows;
}
int32_t mgmtGetVnodeMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetVnodeMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SUserObj *pUser = mgmtGetUserFromConn(pConn);
if (pUser == NULL) return 0;
@ -571,11 +571,11 @@ void *mgmtGetNextDnodeImp(SShowObj *pShow, SDnodeObj **pDnode) {
void *(*mgmtGetNextDnode)(SShowObj *pShow, SDnodeObj **pDnode) = mgmtGetNextDnodeImp;
int32_t mgmtGetScoresMetaImp(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetScoresMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
int32_t (*mgmtGetScoresMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetScoresMetaImp;
int32_t (*mgmtGetScoresMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetScoresMetaImp;
int32_t mgmtRetrieveScoresImp(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
return 0;

6
src/mnode/src/mgmtDnodeInt.c
View File

@ -83,7 +83,7 @@ int32_t mgmtProcessMeterCfgMsg(int8_t *pCont, int32_t contLen, void *pConn) {
return TSDB_CODE_REDIRECT;
}
SMeterCfgMsg *cfg = (SMeterCfgMsg *) pConn;
STableCfgMsg *cfg = (STableCfgMsg *) pConn;
int32_t vnode = htonl(cfg->vnode);
int32_t sid = htonl(cfg->sid);
@ -306,7 +306,7 @@ int mgmtSendRemoveMeterMsgToDnode(STableInfo *pTable, SVgObj *pVgroup) {
// pRemove = (SDRemoveTableMsg *)pMsg;
// pRemove->vnode = htons(pVgroup->vnodeGid[i].vnode);
// pRemove->sid = htonl(pTable->gid.sid);
// memcpy(pRemove->meterId, pTable->meterId, TSDB_TABLE_ID_LEN);
// memcpy(pRemove->tableId, pTable->tableId, TSDB_TABLE_ID_LEN);
//
// pMsg += sizeof(SDRemoveTableMsg);
// msgLen = pMsg - pStart;
@ -372,7 +372,7 @@ char *mgmtBuildVpeersIe(char *pMsg, SVgObj *pVgroup, int vnode) {
pCfg->daysToKeep2 = htonl(pCfg->daysToKeep2);
pCfg->daysToKeep = htonl(pCfg->daysToKeep);
pCfg->commitTime = htonl(pCfg->commitTime);
pCfg->blocksPerMeter = htons(pCfg->blocksPerMeter);
pCfg->blocksPerTable = htons(pCfg->blocksPerTable);
pCfg->replications = (char)pVgroup->numOfVnodes;
pCfg->rowsInFileBlock = htonl(pCfg->rowsInFileBlock);

4
src/mnode/src/mgmtGrant.c
View File

@ -37,8 +37,8 @@ int32_t (*mgmtCheckTimeSeries)(uint32_t timeseries) = mgmtCheckTimeSeriesImp;
bool mgmtCheckExpiredImp() { return false; }
bool (*mgmtCheckExpired)() = mgmtCheckExpiredImp;
int32_t mgmtGetGrantsMetaImp(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) { return TSDB_CODE_OPS_NOT_SUPPORT; }
int32_t (*mgmtGetGrantsMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetGrantsMetaImp;
int32_t mgmtGetGrantsMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) { return TSDB_CODE_OPS_NOT_SUPPORT; }
int32_t (*mgmtGetGrantsMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetGrantsMetaImp;
int32_t mgmtRetrieveGrantsImp(SShowObj *pShow, char *data, int rows, void *pConn) { return 0; }
int32_t (*mgmtRetrieveGrants)(SShowObj *pShow, char *data, int rows, void *pConn) = mgmtRetrieveGrantsImp;

4
src/mnode/src/mgmtMnode.c
View File

@ -16,11 +16,11 @@
#define _DEFAULT_SOURCE
#include "mgmtMnode.h"
int32_t mgmtGetMnodeMetaImp(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetMnodeMetaImp(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
return TSDB_CODE_OPS_NOT_SUPPORT;
}
int32_t (*mgmtGetMnodeMeta)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetMnodeMetaImp;
int32_t (*mgmtGetMnodeMeta)(STableMeta *pMeta, SShowObj *pShow, void *pConn) = mgmtGetMnodeMetaImp;
int32_t mgmtRetrieveMnodesImp(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
return 0;

4
src/mnode/src/mgmtNormalTable.c
View File

@ -456,7 +456,7 @@ static int32_t mgmtSetSchemaFromNormalTable(SSchema *pSchema, SNormalTableObj *p
return numOfCols * sizeof(SSchema);
}
int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp) {
int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, STableMeta *pMeta, bool usePublicIp) {
pMeta->uid = htobe64(pTable->uid);
pMeta->sid = htonl(pTable->sid);
pMeta->vgid = htonl(pTable->vgId);
@ -465,7 +465,7 @@ int32_t mgmtGetNormalTableMeta(SDbObj *pDb, SNormalTableObj *pTable, SMeterMeta
pMeta->numOfTags = 0;
pMeta->numOfColumns = htons(pTable->numOfColumns);
pMeta->tableType = pTable->type;
pMeta->contLen = sizeof(SMeterMeta) + mgmtSetSchemaFromNormalTable(pMeta->schema, pTable);
pMeta->contLen = sizeof(STableMeta) + mgmtSetSchemaFromNormalTable(pMeta->schema, pTable);
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {

4
src/mnode/src/mgmtProfile.c
View File

@ -117,7 +117,7 @@ int32_t mgmtGetQueries(SShowObj *pShow, void *pConn) {
return 0;
}
int32_t mgmtGetQueryMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetQueryMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SSchema *pSchema = tsGetSchema(pMeta);
@ -316,7 +316,7 @@ int32_t mgmtGetStreams(SShowObj *pShow, void *pConn) {
return 0;
}
int32_t mgmtGetStreamMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetStreamMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t cols = 0;
SSchema *pSchema = tsGetSchema(pMeta);

28
src/mnode/src/mgmtShell.c
View File

@ -40,9 +40,9 @@
#include "mgmtVgroup.h"
#define MAX_LEN_OF_METER_META (sizeof(SMultiMeterMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS + sizeof(SSchema) * TSDB_MAX_TAGS + TSDB_MAX_TAGS_LEN)
#define MAX_LEN_OF_METER_META (sizeof(SMultiTableMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS + sizeof(SSchema) * TSDB_MAX_TAGS + TSDB_MAX_TAGS_LEN)
typedef int32_t (*GetMateFp)(SMeterMeta *pMeta, SShowObj *pShow, void *pConn);
typedef int32_t (*GetMateFp)(STableMeta *pMeta, SShowObj *pShow, void *pConn);
typedef int32_t (*RetrieveMetaFp)(SShowObj *pShow, char *data, int32_t rows, void *pConn);
static GetMateFp* mgmtGetMetaFp;
static RetrieveMetaFp* mgmtRetrieveFp;
@ -118,7 +118,7 @@ void mgmtCleanUpShell() {
}
}
int32_t mgmtProcessMeterMetaMsg(void *pCont, int32_t contLen, void *ahandle) {
int32_t mgmtProcessTableMetaMsg(void *pCont, int32_t contLen, void *ahandle) {
SRpcConnInfo connInfo;
rpcGetConnInfo(ahandle, &connInfo);
@ -177,7 +177,7 @@ int32_t mgmtProcessMeterMetaMsg(void *pCont, int32_t contLen, void *ahandle) {
return TSDB_CODE_INVALID_TABLE;
}
SMeterMeta *pMeta = rpcMallocCont(sizeof(SMeterMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS);
STableMeta *pMeta = rpcMallocCont(sizeof(STableMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS);
int32_t code = mgmtGetTableMeta(pDb, pTable, pMeta, usePublicIp);
if (code == TSDB_CODE_SUCCESS) {
@ -206,13 +206,13 @@ int32_t mgmtProcessMultiMeterMetaMsg(void *pCont, int32_t contLen, void *ahandle
pInfo->numOfTables = htonl(pInfo->numOfTables);
int32_t totalMallocLen = 4*1024*1024; // first malloc 4 MB, subsequent reallocation as twice
SMultiMeterMeta *pMultiMeta = rpcMallocCont(totalMallocLen);
SMultiTableMeta *pMultiMeta = rpcMallocCont(totalMallocLen);
if (pMultiMeta == NULL) {
rpcSendResponse(ahandle, TSDB_CODE_SERV_OUT_OF_MEMORY, NULL, 0);
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
pMultiMeta->contLen = sizeof(SMultiMeterMeta);
pMultiMeta->contLen = sizeof(SMultiTableMeta);
pMultiMeta->numOfTables = 0;
for (int t = 0; t < pInfo->numOfTables; ++t) {
@ -224,7 +224,7 @@ int32_t mgmtProcessMultiMeterMetaMsg(void *pCont, int32_t contLen, void *ahandle
if (pDb == NULL) continue;
int availLen = totalMallocLen - pMultiMeta->contLen;
if (availLen <= sizeof(SMeterMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS) {
if (availLen <= sizeof(STableMeta) + sizeof(SSchema) * TSDB_MAX_COLUMNS) {
//TODO realloc
//totalMallocLen *= 2;
//pMultiMeta = rpcReMalloc(pMultiMeta, totalMallocLen);
@ -237,7 +237,7 @@ int32_t mgmtProcessMultiMeterMetaMsg(void *pCont, int32_t contLen, void *ahandle
//}
}
SMeterMeta *pMeta = (SMeterMeta *)(pMultiMeta->metas + pMultiMeta->contLen);
STableMeta *pMeta = (STableMeta *)(pMultiMeta->metas + pMultiMeta->contLen);
int32_t code = mgmtGetTableMeta(pDb, pTable, pMeta, usePublicIp);
if (code == TSDB_CODE_SUCCESS) {
pMultiMeta->numOfTables ++;
@ -269,7 +269,7 @@ int32_t mgmtProcessCreateDbMsg(void *pCont, int32_t contLen, void *ahandle) {
pCreate->daysToKeep1 = htonl(pCreate->daysToKeep1);
pCreate->daysToKeep2 = htonl(pCreate->daysToKeep2);
pCreate->commitTime = htonl(pCreate->commitTime);
pCreate->blocksPerMeter = htons(pCreate->blocksPerMeter);
pCreate->blocksPerTable = htons(pCreate->blocksPerTable);
pCreate->rowsInFileBlock = htonl(pCreate->rowsInFileBlock);
// pCreate->cacheNumOfBlocks = htonl(pCreate->cacheNumOfBlocks);
@ -673,7 +673,7 @@ int32_t mgmtProcessShowMsg(void *pCont, int32_t contLen, void *ahandle) {
mgmtSaveQhandle(pShow);
pShowRsp->qhandle = htobe64((uint64_t) pShow);
code = (*mgmtGetMetaFp[(uint8_t) pShowMsg->type])(&pShowRsp->meterMeta, pShow, ahandle);
code = (*mgmtGetMetaFp[(uint8_t) pShowMsg->type])(&pShowRsp->tableMeta, pShow, ahandle);
if (code == 0) {
size = sizeof(SShowRsp) + sizeof(SSchema) * pShow->numOfColumns;
} else {
@ -691,7 +691,7 @@ int32_t mgmtProcessRetrieveMsg(void *pCont, int32_t contLen, void *ahandle) {
int32_t rowsToRead = 0;
int32_t size = 0;
int32_t rowsRead = 0;
SRetrieveMeterMsg *pRetrieve = (SRetrieveMeterMsg *)pCont;
SRetrieveTableMsg *pRetrieve = (SRetrieveTableMsg *)pCont;
pRetrieve->qhandle = htobe64(pRetrieve->qhandle);
/*
@ -726,7 +726,7 @@ int32_t mgmtProcessRetrieveMsg(void *pCont, int32_t contLen, void *ahandle) {
}
size += 100;
SRetrieveMeterRsp *pRsp = rpcMallocCont(size);
SRetrieveTableRsp *pRsp = rpcMallocCont(size);
// if free flag is set, client wants to clean the resources
if ((pRetrieve->free & TSDB_QUERY_TYPE_FREE_RESOURCE) != TSDB_QUERY_TYPE_FREE_RESOURCE)
@ -1040,7 +1040,7 @@ connect_over:
}
/**
* check if we need to add mgmtProcessMeterMetaMsg into tranQueue, which will be executed one-by-one.
* check if we need to add mgmtProcessTableMetaMsg into tranQueue, which will be executed one-by-one.
*/
static bool mgmtCheckMeterMetaMsgType(void *pMsg) {
STableInfoMsg *pInfo = (STableInfoMsg *) pMsg;
@ -1114,7 +1114,7 @@ void mgmtInitProcessShellMsg() {
mgmtProcessShellMsg[TSDB_MSG_TYPE_KILL_CONNECTION] = mgmtProcessKillConnectionMsg;
mgmtProcessShellMsg[TSDB_MSG_TYPE_SHOW] = mgmtProcessShowMsg;
mgmtProcessShellMsg[TSDB_MSG_TYPE_RETRIEVE] = mgmtProcessRetrieveMsg;
mgmtProcessShellMsg[TSDB_MSG_TYPE_TABLE_META] = mgmtProcessMeterMetaMsg;
mgmtProcessShellMsg[TSDB_MSG_TYPE_TABLE_META] = mgmtProcessTableMetaMsg;
mgmtProcessShellMsg[TSDB_MSG_TYPE_MULTI_TABLE_META] = mgmtProcessMultiMeterMetaMsg;
mgmtProcessShellMsg[TSDB_MSG_TYPE_STABLE_META] = mgmtProcessUnSupportMsg;
}

4
src/mnode/src/mgmtStreamTable.c
View File

@ -396,7 +396,7 @@ static int32_t mgmtSetSchemaFromStreamTable(SSchema *pSchema, SStreamTableObj *p
return numOfCols * sizeof(SSchema);
}
int32_t mgmtGetStreamTableMeta(SDbObj *pDb, SStreamTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp) {
int32_t mgmtGetStreamTableMeta(SDbObj *pDb, SStreamTableObj *pTable, STableMeta *pMeta, bool usePublicIp) {
pMeta->uid = htobe64(pTable->uid);
pMeta->sid = htonl(pTable->sid);
pMeta->vgid = htonl(pTable->vgId);
@ -405,7 +405,7 @@ int32_t mgmtGetStreamTableMeta(SDbObj *pDb, SStreamTableObj *pTable, SMeterMeta
pMeta->numOfTags = 0;
pMeta->numOfColumns = htons(pTable->numOfColumns);
pMeta->tableType = pTable->type;
pMeta->contLen = sizeof(SMeterMeta) + mgmtSetSchemaFromStreamTable(pMeta->schema, pTable);
pMeta->contLen = sizeof(STableMeta) + mgmtSetSchemaFromStreamTable(pMeta->schema, pTable);
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {

6
src/mnode/src/mgmtSuperTable.c
View File

@ -457,7 +457,7 @@ int32_t mgmtDropSuperTableColumnByName(SSuperTableObj *pStable, char *colName) {
return TSDB_CODE_SUCCESS;
}
int32_t mgmtGetShowSuperTableMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetShowSuperTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
// int32_t cols = 0;
//
// SDbObj *pDb = NULL;
@ -616,7 +616,7 @@ int32_t mgmtSetSchemaFromSuperTable(SSchema *pSchema, SSuperTableObj *pTable) {
return (pTable->numOfColumns + pTable->numOfTags) * sizeof(SSchema);
}
int32_t mgmtGetSuperTableMeta(SDbObj *pDb, SSuperTableObj *pTable, SMeterMeta *pMeta, bool usePublicIp) {
int32_t mgmtGetSuperTableMeta(SDbObj *pDb, SSuperTableObj *pTable, STableMeta *pMeta, bool usePublicIp) {
pMeta->uid = htobe64(pTable->uid);
pMeta->sid = htonl(pTable->sid);
pMeta->vgid = htonl(pTable->vgId);
@ -625,7 +625,7 @@ int32_t mgmtGetSuperTableMeta(SDbObj *pDb, SSuperTableObj *pTable, SMeterMeta *p
pMeta->numOfTags = pTable->numOfTags;
pMeta->numOfColumns = htons(pTable->numOfColumns);
pMeta->tableType = pTable->type;
pMeta->contLen = sizeof(SMeterMeta) + mgmtSetSchemaFromSuperTable(pMeta->schema, pTable);
pMeta->contLen = sizeof(STableMeta) + mgmtSetSchemaFromSuperTable(pMeta->schema, pTable);
SVgObj *pVgroup = mgmtGetVgroup(pTable->vgId);
if (pVgroup == NULL) {

4
src/mnode/src/mgmtTable.c
View File

@ -99,7 +99,7 @@ STableInfo* mgmtGetTableByPos(uint32_t dnodeIp, int32_t vnode, int32_t sid) {
return NULL;
}
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, SMeterMeta *pMeta, bool usePublicIp) {
int32_t mgmtGetTableMeta(SDbObj *pDb, STableInfo *pTable, STableMeta *pMeta, bool usePublicIp) {
if (pTable->type == TSDB_TABLE_TYPE_CHILD_TABLE) {
mgmtGetChildTableMeta(pDb, (SChildTableObj *) pTable, pMeta, usePublicIp);
} else if (pTable->type == TSDB_TABLE_TYPE_STREAM_TABLE) {
@ -247,7 +247,7 @@ void mgmtCleanUpMeters() {
mgmtCleanUpSuperTables();
}
int32_t mgmtGetShowTableMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetShowTableMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
// int32_t cols = 0;
//
// SDbObj *pDb = NULL;

2
src/mnode/src/mgmtUser.c
View File

@ -158,7 +158,7 @@ void mgmtCleanUpUsers() {
sdbCloseTable(tsUserSdb);
}
int32_t mgmtGetUserMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetUserMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
// int32_t cols = 0;
// SSchema *pSchema = tsGetSchema(pMeta);
//

2
src/mnode/src/mgmtVgroup.c
View File

@ -248,7 +248,7 @@ void mgmtSetVgroupIdPool() {
void mgmtCleanUpVgroups() { sdbCloseTable(tsVgroupSdb); }
int32_t mgmtGetVgroupMeta(SMeterMeta *pMeta, SShowObj *pShow, void *pConn) {
int32_t mgmtGetVgroupMeta(STableMeta *pMeta, SShowObj *pShow, void *pConn) {
// int32_t cols = 0;
//
// SDbObj *pDb = NULL;

2
src/util/inc/textbuffer.h
View File

@ -124,7 +124,7 @@ typedef struct tTagSchema {
typedef struct tSidSet {
int32_t numOfSids;
int32_t numOfSubSet;
SMeterSidExtInfo **pSids;
STableSidExtInfo **pSids;
int32_t * starterPos; // position of each subgroup, generated according to
SColumnModel *pColumnModel;

14
src/util/inc/tschemautil.h
View File

@ -42,20 +42,20 @@ struct SSchema;
*/
bool isValidSchema(struct SSchema *pSchema, int32_t numOfCols);
struct SSchema *tsGetSchema(SMeterMeta *pMeta);
struct SSchema *tsGetSchema(STableMeta *pMeta);
struct SSchema *tsGetTagSchema(SMeterMeta *pMeta);
struct SSchema *tsGetTagSchema(STableMeta *pMeta);
struct SSchema *tsGetColumnSchema(SMeterMeta *pMeta, int32_t startCol);
struct SSchema *tsGetColumnSchema(STableMeta *pMeta, int32_t startCol);
struct SSchema tsGetTbnameColumnSchema();
char *tsGetTagsValue(SMeterMeta *pMeta);
char *tsGetTagsValue(STableMeta *pMeta);
bool tsMeterMetaIdentical(SMeterMeta *p1, SMeterMeta *p2);
bool tsMeterMetaIdentical(STableMeta *p1, STableMeta *p2);
void extractTableName(char *meterId, char *name);
void extractTableName(char *tableId, char *name);
SSQLToken extractDBName(char *meterId, char *name);
SSQLToken extractDBName(char *tableId, char *name);
void extractTableNameFromToken(SSQLToken *pToken, SSQLToken* pTable);