Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
未验证 提交 a23418d1 编写于 作者: D dapan1121 提交者： GitHub
浏览文件
操作

Merge pull request #14070 from taosdata/feature/showcreate 

feat: show create table
上级 668e196b 4bbec6ee
展开全部 隐藏空白更改
内联 并排
Showing with 2286 addition and 274 deletion
include/common/tmsg.h
浏览文件 @ a23418d1
......@@ -168,7 +168,7 @@ typedef struct {
int32_t vgId;
char* dbFName;
char* tbName;
} SBuildTableMetaInput;
} SBuildTableInput;
typedef struct {
char db[TSDB_DB_FNAME_LEN];
......@@ -444,6 +444,7 @@ typedef struct {
char* comment;
char* pAst1;
char* pAst2;
SArray* pFuncs;
} SMCreateStbReq;
int32_t tSerializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
......@@ -667,6 +668,41 @@ int32_t tSerializeSQueryTableRsp(void* buf, int32_t bufLen, SQueryTableRsp* pRsp
int32_t tDeserializeSQueryTableRsp(void* buf, int32_t bufLen, SQueryTableRsp* pRsp);
typedef struct {
SMsgHead header;
char dbFName[TSDB_DB_FNAME_LEN];
char tbName[TSDB_TABLE_NAME_LEN];
} STableCfgReq;
typedef struct {
char tbName[TSDB_TABLE_NAME_LEN];
char stbName[TSDB_TABLE_NAME_LEN];
char dbFName[TSDB_DB_FNAME_LEN];
int32_t numOfTags;
int32_t numOfColumns;
int8_t tableType;
int64_t delay1;
int64_t delay2;
int64_t watermark1;
int64_t watermark2;
int32_t ttl;
SArray* pFuncs;
int32_t commentLen;
char* pComment;
SSchema* pSchemas;
int32_t tagsLen;
char* pTags;
} STableCfg;
typedef STableCfg STableCfgRsp;
int32_t tSerializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq);
int32_t tDeserializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq);
int32_t tSerializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp);
int32_t tDeserializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp);
void tFreeSTableCfgRsp(STableCfgRsp *pRsp);
typedef struct {
char db[TSDB_DB_FNAME_LEN];
int32_t numOfVgroups;
......
include/common/tmsgdef.h
浏览文件 @ a23418d1
......@@ -131,6 +131,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_MND_DROP_INDEX, "drop-index", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_GET_INDEX, "get-index", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_GET_TABLE_INDEX, "get-table-index", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_TABLE_CFG, "table-cfg", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "create-topic", SMCreateTopicReq, SMCreateTopicRsp)
TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "alter-topic", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "drop-topic", NULL, NULL)
......@@ -171,6 +172,7 @@ enum {
TD_DEF_MSG_TYPE(TDMT_VND_UPDATE_TAG_VAL, "update-tag-val", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_TABLE_META, "vnode-table-meta", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_TABLES_META, "vnode-tables-meta", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_TABLE_CFG, "vnode-table-cfg", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_CREATE_STB, "vnode-create-stb", SVCreateStbReq, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ALTER_STB, "vnode-alter-stb", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_DROP_STB, "vnode-drop-stb", SVDropStbReq, NULL)
......
include/common/tname.h
浏览文件 @ a23418d1
......@@ -63,6 +63,8 @@ int32_t tNameSetAcctId(SName* dst, int32_t acctId);
bool tNameDBNameEqual(SName* left, SName* right);
bool tNameTbNameEqual(SName* left, SName* right);
typedef struct {
// input
SArray* tags; // element is SSmlKv
......
include/libs/catalog/catalog.h
浏览文件 @ a23418d1
......@@ -68,6 +68,7 @@ typedef struct SCatalogReq {
SArray* pIndex; // element is index name
SArray* pUser; // element is SUserAuthInfo
SArray* pTableIndex; // element is SNAME
SArray* pTableCfg; // element is SNAME
bool qNodeRequired; // valid qnode
bool dNodeRequired; // valid dnode
bool forceUpdate;
......@@ -89,6 +90,7 @@ typedef struct SMetaData {
SArray* pIndex; // pRes = SIndexInfo*
SArray* pUser; // pRes = bool*
SArray* pQnodeList; // pRes = SArray<SQueryNodeLoad>*
SArray* pTableCfg; // pRes = STableCfg*
SArray* pDnodeList; // pRes = SArray<SEpSet>*
} SMetaData;
......@@ -284,6 +286,8 @@ int32_t catalogGetIndexMeta(SCatalog* pCtg, SRequestConnInfo* pConn, const char*
int32_t catalogGetTableIndex(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTableName, SArray** pRes);
int32_t catalogRefreshGetTableCfg(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg** pCfg);
int32_t catalogUpdateTableIndex(SCatalog* pCtg, STableIndexRsp *pRsp);
int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo* pConn, const char* funcName, SFuncInfo* pInfo);
......
include/libs/nodes/cmdnodes.h
浏览文件 @ a23418d1
......@@ -28,6 +28,15 @@ extern "C" {
#define DESCRIBE_RESULT_TYPE_LEN (20 + VARSTR_HEADER_SIZE)
#define DESCRIBE_RESULT_NOTE_LEN (8 + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_DB_RESULT_COLS 2
#define SHOW_CREATE_DB_RESULT_FIELD1_LEN (TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_DB_RESULT_FIELD2_LEN (TSDB_MAX_BINARY_LEN + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_TB_RESULT_COLS 2
#define SHOW_CREATE_TB_RESULT_FIELD1_LEN (TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE)
#define SHOW_CREATE_TB_RESULT_FIELD2_LEN (TSDB_MAX_BINARY_LEN + VARSTR_HEADER_SIZE)
#define PRIVILEGE_TYPE_MASK(n) (1 << n)
#define PRIVILEGE_TYPE_ALL PRIVILEGE_TYPE_MASK(0)
......@@ -221,7 +230,7 @@ typedef struct SShowCreateTableStmt {
ENodeType type;
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
STableMeta* pMeta;
void* pCfg; // STableCfg
} SShowCreateTableStmt;
typedef struct SShowTableDistributedStmt {
......
include/libs/qcom/query.h
浏览文件 @ a23418d1
......@@ -207,6 +207,10 @@ char* jobTaskStatusStr(int32_t status);
SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name);
void destroyQueryExecRes(SQueryExecRes* pRes);
int32_t dataConverToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len);
char* parseTagDatatoJson(void* p);
int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst);
int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst);
extern int32_t (*queryBuildMsg[TDMT_MAX])(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallocFp)(int32_t));
extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t msgSize);
......
include/util/taoserror.h
浏览文件 @ a23418d1
......@@ -129,6 +129,7 @@ int32_t* taosGetErrno();
#define TSDB_CODE_TSC_STMT_CLAUSE_ERROR TAOS_DEF_ERROR_CODE(0, 0X0227)
#define TSDB_CODE_TSC_QUERY_KILLED TAOS_DEF_ERROR_CODE(0, 0X0228)
#define TSDB_CODE_TSC_NO_EXEC_NODE TAOS_DEF_ERROR_CODE(0, 0X0229)
#define TSDB_CODE_TSC_NOT_STABLE_ERROR TAOS_DEF_ERROR_CODE(0, 0X022a)
// mnode-common
#define TSDB_CODE_MND_APP_ERROR TAOS_DEF_ERROR_CODE(0, 0x0300)
......
source/client/src/clientImpl.c
浏览文件 @ a23418d1
......@@ -244,10 +244,10 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
}
SReqResultInfo* pResultInfo = &pRequest->body.resInfo;
pRequest->code = code;
if (pRequest->code != TSDB_CODE_SUCCESS) {
pResultInfo->numOfRows = 0;
pRequest->code = code;
tscError("0x%" PRIx64 " fetch results failed, code:%s, reqId:0x%" PRIx64, pRequest->self, tstrerror(code),
pRequest->requestId);
} else {
......@@ -256,7 +256,7 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
pRequest->requestId);
}
pRequest->body.queryFp(pRequest->body.param, pRequest, 0);
pRequest->body.queryFp(pRequest->body.param, pRequest, code);
// pRequest->body.fetchFp(pRequest->body.param, pRequest, pResultInfo->numOfRows);
}
......@@ -1444,80 +1444,6 @@ static int32_t doPrepareResPtr(SReqResultInfo* pResInfo) {
return TSDB_CODE_SUCCESS;
}
static char* parseTagDatatoJson(void* p) {
char* string = NULL;
cJSON* json = cJSON_CreateObject();
if (json == NULL) {
goto end;
}
SArray* pTagVals = NULL;
if (tTagToValArray((const STag*)p, &pTagVals) != 0) {
goto end;
}
int16_t nCols = taosArrayGetSize(pTagVals);
char tagJsonKey[256] = {0};
for (int j = 0; j < nCols; ++j) {
STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);
// json key encode by binary
memset(tagJsonKey, 0, sizeof(tagJsonKey));
memcpy(tagJsonKey, pTagVal->pKey, strlen(pTagVal->pKey));
// json value
char type = pTagVal->type;
if (type == TSDB_DATA_TYPE_NULL) {
cJSON* value = cJSON_CreateNull();
if (value == NULL) {
goto end;
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else if (type == TSDB_DATA_TYPE_NCHAR) {
cJSON* value = NULL;
if (pTagVal->nData > 0) {
char* tagJsonValue = taosMemoryCalloc(pTagVal->nData, 1);
int32_t length = taosUcs4ToMbs((TdUcs4*)pTagVal->pData, pTagVal->nData, tagJsonValue);
if (length < 0) {
tscError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset,
pTagVal->pData);
taosMemoryFree(tagJsonValue);
goto end;
}
value = cJSON_CreateString(tagJsonValue);
taosMemoryFree(tagJsonValue);
if (value == NULL) {
goto end;
}
} else if (pTagVal->nData == 0) {
value = cJSON_CreateString("");
} else {
ASSERT(0);
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else if (type == TSDB_DATA_TYPE_DOUBLE) {
double jsonVd = *(double*)(&pTagVal->i64);
cJSON* value = cJSON_CreateNumber(jsonVd);
if (value == NULL) {
goto end;
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else if (type == TSDB_DATA_TYPE_BOOL) {
char jsonVd = *(char*)(&pTagVal->i64);
cJSON* value = cJSON_CreateBool(jsonVd);
if (value == NULL) {
goto end;
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else {
ASSERT(0);
}
}
string = cJSON_PrintUnformatted(json);
end:
cJSON_Delete(json);
return string;
}
static int32_t doConvertUCS4(SReqResultInfo* pResultInfo, int32_t numOfRows, int32_t numOfCols, int32_t* colLength) {
for (int32_t i = 0; i < numOfCols; ++i) {
int32_t type = pResultInfo->fields[i].type;
......
source/common/src/tmsg.c
浏览文件 @ a23418d1
......@@ -532,6 +532,14 @@ int32_t tSerializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pReq
if (pReq->ast2Len > 0) {
if (tEncodeBinary(&encoder, pReq->pAst2, pReq->ast2Len) < 0) return -1;
}
int32_t numOfFuncs = taosArrayGetSize(pReq->pFuncs);
if (tEncodeI32(&encoder, numOfFuncs) < 0) return -1;
for (int32_t i = 0; i < numOfFuncs; ++i) {
const char *pFunc = taosArrayGet(pReq->pFuncs, i);
if (tEncodeCStr(&encoder, pFunc) < 0) return -1;
}
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
......@@ -606,6 +614,21 @@ int32_t tDeserializeSMCreateStbReq(void *buf, int32_t bufLen, SMCreateStbReq *pR
if (tDecodeCStrTo(&decoder, pReq->pAst2) < 0) return -1;
}
int32_t numOfFuncs = 0;
if (tDecodeI32(&decoder, &numOfFuncs) < 0) return -1;
if (numOfFuncs > 0) {
pReq->pFuncs = taosArrayInit(numOfFuncs, TSDB_FUNC_NAME_LEN);
if (NULL == pReq->pFuncs) return -1;
}
for (int32_t i = 0; i < numOfFuncs; ++i) {
char* pFunc = NULL;
if (tDecodeCStrAlloc(&decoder, &pFunc) < 0) return -1;
if (taosArrayPush(pReq->pFuncs, pFunc) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
tEndDecode(&decoder);
tDecoderClear(&decoder);
......@@ -618,8 +641,7 @@ void tFreeSMCreateStbReq(SMCreateStbReq *pReq) {
taosMemoryFreeClear(pReq->comment);
taosMemoryFreeClear(pReq->pAst1);
taosMemoryFreeClear(pReq->pAst2);
pReq->pColumns = NULL;
pReq->pTags = NULL;
taosArrayDestroy(pReq->pFuncs);
}
int32_t tSerializeSMDropStbReq(void *buf, int32_t bufLen, SMDropStbReq *pReq) {
......@@ -1757,6 +1779,165 @@ void tFreeSRetrieveFuncRsp(SRetrieveFuncRsp *pRsp) {
taosArrayDestroy(pRsp->pFuncInfos);
}
int32_t tSerializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
if (buf != NULL) {
buf = (char *)buf + headLen;
bufLen -= headLen;
}
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->dbFName) < 0) return -1;
if (tEncodeCStr(&encoder, pReq->tbName) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
if (buf != NULL) {
SMsgHead *pHead = (SMsgHead *)((char *)buf - headLen);
pHead->vgId = htonl(pReq->header.vgId);
pHead->contLen = htonl(tlen + headLen);
}
return tlen + headLen;
}
int32_t tDeserializeSTableCfgReq(void *buf, int32_t bufLen, STableCfgReq *pReq) {
int32_t headLen = sizeof(SMsgHead);
SMsgHead *pHead = buf;
pHead->vgId = pReq->header.vgId;
pHead->contLen = pReq->header.contLen;
SDecoder decoder = {0};
tDecoderInit(&decoder, (char *)buf + headLen, bufLen - headLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->dbFName) < 0) return -1;
if (tDecodeCStrTo(&decoder, pReq->tbName) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
int32_t tSerializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
if (tStartEncode(&encoder) < 0) return -1;
if (tEncodeCStr(&encoder, pRsp->tbName) < 0) return -1;
if (tEncodeCStr(&encoder, pRsp->stbName) < 0) return -1;
if (tEncodeCStr(&encoder, pRsp->dbFName) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->numOfTags) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->numOfColumns) < 0) return -1;
if (tEncodeI8(&encoder, pRsp->tableType) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->delay1) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->delay2) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->watermark1) < 0) return -1;
if (tEncodeI64(&encoder, pRsp->watermark2) < 0) return -1;
if (tEncodeI32(&encoder, pRsp->ttl) < 0) return -1;
int32_t numOfFuncs = taosArrayGetSize(pRsp->pFuncs);
if (tEncodeI32(&encoder, numOfFuncs) < 0) return -1;
for (int32_t i = 0; i < numOfFuncs; ++i) {
const char *pFunc = taosArrayGet(pRsp->pFuncs, i);
if (tEncodeCStr(&encoder, pFunc) < 0) return -1;
}
if (tEncodeI32(&encoder, pRsp->commentLen) < 0) return -1;
if (pRsp->commentLen > 0) {
if (tEncodeCStr(&encoder, pRsp->pComment) < 0) return -1;
}
for (int32_t i = 0; i < pRsp->numOfColumns + pRsp->numOfTags; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
if (tEncodeSSchema(&encoder, pSchema) < 0) return -1;
}
if (tEncodeI32(&encoder, pRsp->tagsLen) < 0) return -1;
if (tEncodeBinary(&encoder, pRsp->pTags, pRsp->tagsLen) < 0) return -1;
tEndEncode(&encoder);
int32_t tlen = encoder.pos;
tEncoderClear(&encoder);
return tlen;
}
int32_t tDeserializeSTableCfgRsp(void *buf, int32_t bufLen, STableCfgRsp *pRsp) {
SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen);
if (tStartDecode(&decoder) < 0) return -1;
if (tDecodeCStrTo(&decoder, pRsp->tbName) < 0) return -1;
if (tDecodeCStrTo(&decoder, pRsp->stbName) < 0) return -1;
if (tDecodeCStrTo(&decoder, pRsp->dbFName) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->numOfTags) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->numOfColumns) < 0) return -1;
if (tDecodeI8(&decoder, &pRsp->tableType) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->delay1) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->delay2) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->watermark1) < 0) return -1;
if (tDecodeI64(&decoder, &pRsp->watermark2) < 0) return -1;
if (tDecodeI32(&decoder, &pRsp->ttl) < 0) return -1;
int32_t numOfFuncs = 0;
if (tDecodeI32(&decoder, &numOfFuncs) < 0) return -1;
if (numOfFuncs > 0) {
pRsp->pFuncs = taosArrayInit(numOfFuncs, TSDB_FUNC_NAME_LEN);
if (NULL == pRsp->pFuncs) return -1;
}
for (int32_t i = 0; i < numOfFuncs; ++i) {
char pFunc[TSDB_FUNC_NAME_LEN];
if (tDecodeCStrTo(&decoder, pFunc) < 0) return -1;
if (taosArrayPush(pRsp->pFuncs, pFunc) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
}
if (tDecodeI32(&decoder, &pRsp->commentLen) < 0) return -1;
if (pRsp->commentLen > 0) {
if (tDecodeCStrAlloc(&decoder, &pRsp->pComment) < 0) return -1;
} else {
pRsp->pComment = NULL;
}
int32_t totalCols = pRsp->numOfTags + pRsp->numOfColumns;
pRsp->pSchemas = taosMemoryMalloc(sizeof(SSchema) * totalCols);
if (pRsp->pSchemas == NULL) return -1;
for (int32_t i = 0; i < totalCols; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
if (tDecodeSSchema(&decoder, pSchema) < 0) return -1;
}
if (tDecodeI32(&decoder, &pRsp->tagsLen) < 0) return -1;
if (tDecodeBinaryAlloc(&decoder, (void**)&pRsp->pTags, NULL) < 0) return -1;
tEndDecode(&decoder);
tDecoderClear(&decoder);
return 0;
}
void tFreeSTableCfgRsp(STableCfgRsp *pRsp) {
if (NULL == pRsp) {
return;
}
taosMemoryFreeClear(pRsp->pComment);
taosMemoryFreeClear(pRsp->pSchemas);
taosMemoryFreeClear(pRsp->pTags);
taosArrayDestroy(pRsp->pFuncs);
}
int32_t tSerializeSCreateDbReq(void *buf, int32_t bufLen, SCreateDbReq *pReq) {
SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen);
......
source/common/src/tname.c
浏览文件 @ a23418d1
......@@ -240,6 +240,15 @@ bool tNameDBNameEqual(SName* left, SName* right) {
return (0 == strcmp(left->dbname, right->dbname));
}
bool tNameTbNameEqual(SName* left, SName* right) {
bool equal = tNameDBNameEqual(left, right);
if (equal) {
return (0 == strcmp(left->tname, right->tname));
}
return equal;
}
int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
assert(dst != NULL && str != NULL && strlen(str) > 0);
......
source/dnode/mgmt/mgmt_mnode/src/mmHandle.c
浏览文件 @ a23418d1
......@@ -182,6 +182,7 @@ SArray *mmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_STB, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_STB, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_TABLE_META, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_TABLE_CFG, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_SMA, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_SMA, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
......
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
浏览文件 @ a23418d1
......@@ -330,6 +330,7 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_CFG, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
......
source/dnode/mnode/impl/inc/mndDef.h
浏览文件 @ a23418d1
......@@ -349,6 +349,7 @@ typedef struct {
int32_t commentLen;
int32_t ast1Len;
int32_t ast2Len;
SArray* pFuncs;
SSchema* pColumns;
SSchema* pTags;
char* comment;
......
source/dnode/mnode/impl/inc/mndInfoSchema.h
浏览文件 @ a23418d1
......@@ -25,6 +25,7 @@ extern "C" {
int32_t mndInitInfos(SMnode *pMnode);
void mndCleanupInfos(SMnode *pMnode);
int32_t mndBuildInsTableSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp);
int32_t mndBuildInsTableCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp);
#ifdef __cplusplus
}
......
source/dnode/mnode/impl/inc/mndPerfSchema.h
浏览文件 @ a23418d1
......@@ -23,6 +23,7 @@ extern "C" {
#endif
int32_t mndBuildPerfsTableSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp);
int32_t mndBuildPerfsTableCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp);
int32_t mndInitPerfs(SMnode *pMnode);
void mndCleanupPerfs(SMnode *pMnode);
......
source/dnode/mnode/impl/src/mndInfoSchema.c
浏览文件 @ a23418d1
......@@ -90,6 +90,38 @@ int32_t mndBuildInsTableSchema(SMnode *pMnode, const char *dbFName, const char *
return 0;
}
int32_t mndBuildInsTableCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
if (NULL == pMnode->infosMeta) {
terrno = TSDB_CODE_MND_NOT_READY;
return -1;
}
STableMetaRsp *pMeta = taosHashGet(pMnode->infosMeta, tbName, strlen(tbName));
if (NULL == pMeta) {
mError("invalid information schema table name:%s", tbName);
terrno = TSDB_CODE_MND_INVALID_SYS_TABLENAME;
return -1;
}
strcpy(pRsp->tbName, pMeta->tbName);
strcpy(pRsp->stbName, pMeta->stbName);
strcpy(pRsp->dbFName, pMeta->dbFName);
pRsp->numOfTags = pMeta->numOfTags;
pRsp->numOfColumns = pMeta->numOfColumns;
pRsp->tableType = pMeta->tableType;
pRsp->pSchemas = taosMemoryCalloc(pMeta->numOfColumns, sizeof(SSchema));
if (pRsp->pSchemas == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
pRsp->pSchemas = NULL;
return -1;
}
memcpy(pRsp->pSchemas, pMeta->pSchemas, pMeta->numOfColumns * sizeof(SSchema));
return 0;
}
int32_t mndInitInfos(SMnode *pMnode) {
pMnode->infosMeta = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), false, HASH_NO_LOCK);
if (pMnode->infosMeta == NULL) {
......
source/dnode/mnode/impl/src/mndPerfSchema.c
浏览文件 @ a23418d1
......@@ -92,6 +92,37 @@ int32_t mndBuildPerfsTableSchema(SMnode *pMnode, const char *dbFName, const char
return 0;
}
int32_t mndBuildPerfsTableCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
if (NULL == pMnode->perfsMeta) {
terrno = TSDB_CODE_MND_NOT_READY;
return -1;
}
STableMetaRsp *pMeta = taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));
if (NULL == pMeta) {
mError("invalid performance schema table name:%s", tbName);
terrno = TSDB_CODE_MND_INVALID_SYS_TABLENAME;
return -1;
}
strcpy(pRsp->tbName, pMeta->tbName);
strcpy(pRsp->stbName, pMeta->stbName);
strcpy(pRsp->dbFName, pMeta->dbFName);
pRsp->numOfTags = pMeta->numOfTags;
pRsp->numOfColumns = pMeta->numOfColumns;
pRsp->tableType = pMeta->tableType;
pRsp->pSchemas = taosMemoryCalloc(pMeta->numOfColumns, sizeof(SSchema));
if (pRsp->pSchemas == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
pRsp->pSchemas = NULL;
return -1;
}
memcpy(pRsp->pSchemas, pMeta->pSchemas, pMeta->numOfColumns * sizeof(SSchema));
return 0;
}
int32_t mndInitPerfs(SMnode *pMnode) {
pMnode->perfsMeta = taosHashInit(20, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
if (pMnode->perfsMeta == NULL) {
......
source/dnode/mnode/impl/src/mndStb.c
浏览文件 @ a23418d1
......@@ -43,6 +43,7 @@ static int32_t mndProcessDropStbReq(SRpcMsg *pReq);
static int32_t mndProcessTableMetaReq(SRpcMsg *pReq);
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter);
static int32_t mndProcessTableCfgReq(SRpcMsg *pReq);
int32_t mndInitStb(SMnode *pMnode) {
SSdbTable table = {
......@@ -62,6 +63,7 @@ int32_t mndInitStb(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);
mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);
mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb);
......@@ -75,7 +77,7 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags) * sizeof(SSchema) + +pStb->commentLen +
pStb->ast1Len + pStb->ast2Len + STB_RESERVE_SIZE;
pStb->ast1Len + pStb->ast2Len + STB_RESERVE_SIZE + taosArrayGetSize(pStb->pFuncs) * TSDB_FUNC_NAME_LEN;
SSdbRaw *pRaw = sdbAllocRaw(SDB_STB, STB_VER_NUMBER, size);
if (pRaw == NULL) goto _OVER;
......@@ -100,6 +102,13 @@ SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
SDB_SET_INT32(pRaw, dataPos, pStb->ast1Len, _OVER)
SDB_SET_INT32(pRaw, dataPos, pStb->ast2Len, _OVER)
int32_t funcNum = taosArrayGetSize(pStb->pFuncs);
SDB_SET_INT32(pRaw, dataPos, funcNum, _OVER)
for (int32_t i = 0; i < funcNum; ++i) {
char* func = taosArrayGet(pStb->pFuncs, i);
SDB_SET_BINARY(pRaw, dataPos, func, TSDB_FUNC_NAME_LEN, _OVER)
}
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
SSchema *pSchema = &pStb->pColumns[i];
SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)
......@@ -181,6 +190,20 @@ static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
SDB_GET_INT32(pRaw, dataPos, &pStb->ast1Len, _OVER)
SDB_GET_INT32(pRaw, dataPos, &pStb->ast2Len, _OVER)
int32_t funcNum = 0;
SDB_GET_INT32(pRaw, dataPos, &funcNum, _OVER)
if (funcNum > 0) {
pStb->pFuncs = taosArrayInit(funcNum, TSDB_FUNC_NAME_LEN);
if (NULL == pStb->pFuncs) {
goto _OVER;
}
char funcName[TSDB_FUNC_NAME_LEN];
for (int32_t i = 0; i < funcNum; ++i) {
SDB_GET_BINARY(pRaw, dataPos, funcName, TSDB_FUNC_NAME_LEN, _OVER)
taosArrayPush(pStb->pFuncs, funcName);
}
}
pStb->pColumns = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchema));
pStb->pTags = taosMemoryCalloc(pStb->numOfTags, sizeof(SSchema));
if (pStb->pColumns == NULL || pStb->pTags == NULL) {
......@@ -250,6 +273,7 @@ static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
taosMemoryFreeClear(pStb->comment);
taosMemoryFreeClear(pStb->pAst1);
taosMemoryFreeClear(pStb->pAst2);
taosArrayDestroy(pStb->pFuncs);
return 0;
}
......@@ -680,6 +704,9 @@ int32_t mndBuildStbFromReq(SMnode *pMnode, SStbObj *pDst, SMCreateStbReq *pCreat
pDst->numOfColumns = pCreate->numOfColumns;
pDst->numOfTags = pCreate->numOfTags;
pDst->commentLen = pCreate->commentLen;
pDst->pFuncs = pCreate->pFuncs;
pCreate->pFuncs = NULL;
if (pDst->commentLen > 0) {
pDst->comment = taosMemoryCalloc(pDst->commentLen + 1, 1);
if (pDst->comment == NULL) {
......@@ -1277,6 +1304,60 @@ static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbNa
return 0;
}
static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableCfgRsp *pRsp) {
taosRLockLatch(&pStb->lock);
int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
if (pRsp->pSchemas == NULL) {
taosRUnLockLatch(&pStb->lock);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
strcpy(pRsp->dbFName, pStb->db);
strcpy(pRsp->tbName, tbName);
strcpy(pRsp->stbName, tbName);
pRsp->numOfTags = pStb->numOfTags;
pRsp->numOfColumns = pStb->numOfColumns;
pRsp->tableType = TSDB_SUPER_TABLE;
pRsp->delay1 = pStb->maxdelay[0];
pRsp->delay2 = pStb->maxdelay[1];
pRsp->watermark1 = pStb->watermark[0];
pRsp->watermark2 = pStb->watermark[1];
pRsp->ttl = pStb->ttl;
pRsp->commentLen = pStb->commentLen;
if (pStb->commentLen > 0) {
pRsp->pComment = strdup(pStb->comment);
}
for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i];
SSchema *pSrcSchema = &pStb->pColumns[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = pSrcSchema->colId;
pSchema->bytes = pSrcSchema->bytes;
}
for (int32_t i = 0; i < pStb->numOfTags; ++i) {
SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
SSchema *pSrcSchema = &pStb->pTags[i];
memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
pSchema->type = pSrcSchema->type;
pSchema->colId = pSrcSchema->colId;
pSchema->bytes = pSrcSchema->bytes;
}
if (pStb->pFuncs) {
pRsp->pFuncs = taosArrayDup(pStb->pFuncs);
}
taosRUnLockLatch(&pStb->lock);
return 0;
}
static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp,
int32_t *smaVer) {
char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
......@@ -1305,6 +1386,32 @@ static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char
return code;
}
static int32_t mndBuildStbCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
char tbFName[TSDB_TABLE_FNAME_LEN] = {0};
snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
if (pDb == NULL) {
terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
return -1;
}
SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
if (pStb == NULL) {
mndReleaseDb(pMnode, pDb);
terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
return -1;
}
int32_t code = mndBuildStbCfgImp(pDb, pStb, tbName, pRsp);
mndReleaseDb(pMnode, pDb);
mndReleaseStb(pMnode, pStb);
return code;
}
static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, const SMAlterStbReq *pAlter, SStbObj *pObj, void **pCont,
int32_t *pLen) {
int32_t ret;
......@@ -1664,6 +1771,63 @@ _OVER:
return code;
}
static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node;
int32_t code = -1;
STableCfgReq cfgReq = {0};
STableCfgRsp cfgRsp = {0};
if (tDeserializeSTableCfgReq(pReq->pCont, pReq->contLen, &cfgReq) != 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto _OVER;
}
if (0 == strcmp(cfgReq.dbFName, TSDB_INFORMATION_SCHEMA_DB)) {
mDebug("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
if (mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp) != 0) {
goto _OVER;
}
} else if (0 == strcmp(cfgReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {
mDebug("performance_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
if (mndBuildPerfsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp) != 0) {
goto _OVER;
}
} else {
mDebug("stb:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
if (mndBuildStbCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp) != 0) {
goto _OVER;
}
}
int32_t rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);
if (rspLen < 0) {
terrno = TSDB_CODE_INVALID_MSG;
goto _OVER;
}
void *pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
goto _OVER;
}
tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp);
pReq->info.rsp = pRsp;
pReq->info.rspLen = rspLen;
code = 0;
mTrace("%s.%s, cfg is retrieved", cfgReq.dbFName, cfgReq.tbName);
_OVER:
if (code != 0) {
mError("stb:%s.%s, failed to retrieve cfg since %s", cfgReq.dbFName, cfgReq.tbName, terrstr());
}
tFreeSTableCfgRsp(&cfgRsp);
return code;
}
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
int32_t *pRspLen) {
SSTbHbRsp hbRsp = {0};
......
source/dnode/vnode/src/inc/vnd.h
浏览文件 @ a23418d1
......@@ -79,6 +79,7 @@ void vnodeBufPoolReset(SVBufPool* pPool);
int32_t vnodeQueryOpen(SVnode* pVnode);
void vnodeQueryClose(SVnode* pVnode);
int32_t vnodeGetTableMeta(SVnode* pVnode, SRpcMsg* pMsg);
int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg);
// vnodeCommit.c
int32_t vnodeBegin(SVnode* pVnode);
......
source/dnode/vnode/src/vnd/vnodeQuery.c
浏览文件 @ a23418d1
......@@ -124,6 +124,115 @@ _exit:
return TSDB_CODE_SUCCESS;
}
int vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg) {
STableCfgReq cfgReq = {0};
STableCfgRsp cfgRsp = {0};
SMetaReader mer1 = {0};
SMetaReader mer2 = {0};
char tableFName[TSDB_TABLE_FNAME_LEN];
SRpcMsg rpcMsg;
int32_t code = 0;
int32_t rspLen = 0;
void *pRsp = NULL;
SSchemaWrapper schema = {0};
SSchemaWrapper schemaTag = {0};
// decode req
if (tDeserializeSTableCfgReq(pMsg->pCont, pMsg->contLen, &cfgReq) != 0) {
code = TSDB_CODE_INVALID_MSG;
goto _exit;
}
strcpy(cfgRsp.tbName, cfgReq.tbName);
memcpy(cfgRsp.dbFName, cfgReq.dbFName, sizeof(cfgRsp.dbFName));
sprintf(tableFName, "%s.%s", cfgReq.dbFName, cfgReq.tbName);
code = vnodeValidateTableHash(pVnode, tableFName);
if (code) {
goto _exit;
}
// query meta
metaReaderInit(&mer1, pVnode->pMeta, 0);
if (metaGetTableEntryByName(&mer1, cfgReq.tbName) < 0) {
code = terrno;
goto _exit;
}
cfgRsp.tableType = mer1.me.type;
if (mer1.me.type == TSDB_SUPER_TABLE) {
code = TSDB_CODE_VND_HASH_MISMATCH;
goto _exit;
} else if (mer1.me.type == TSDB_CHILD_TABLE) {
metaReaderInit(&mer2, pVnode->pMeta, 0);
if (metaGetTableEntryByUid(&mer2, mer1.me.ctbEntry.suid) < 0) goto _exit;
strcpy(cfgRsp.stbName, mer2.me.name);
schema = mer2.me.stbEntry.schemaRow;
schemaTag = mer2.me.stbEntry.schemaTag;
cfgRsp.ttl = mer1.me.ctbEntry.ttlDays;
cfgRsp.commentLen = mer1.me.ctbEntry.commentLen;
if (mer1.me.ctbEntry.commentLen > 0) {
cfgRsp.pComment = strdup(mer1.me.ctbEntry.comment);
}
STag *pTag = (STag *)mer1.me.ctbEntry.pTags;
cfgRsp.tagsLen = pTag->len;
cfgRsp.pTags = taosMemoryMalloc(cfgRsp.tagsLen);
memcpy(cfgRsp.pTags, pTag, cfgRsp.tagsLen);
} else if (mer1.me.type == TSDB_NORMAL_TABLE) {
schema = mer1.me.ntbEntry.schemaRow;
cfgRsp.ttl = mer1.me.ntbEntry.ttlDays;
cfgRsp.commentLen = mer1.me.ntbEntry.commentLen;
if (mer1.me.ntbEntry.commentLen > 0) {
cfgRsp.pComment = strdup(mer1.me.ntbEntry.comment);
}
} else {
ASSERT(0);
}
cfgRsp.numOfTags = schemaTag.nCols;
cfgRsp.numOfColumns = schema.nCols;
cfgRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (cfgRsp.numOfColumns + cfgRsp.numOfTags));
memcpy(cfgRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);
if (schemaTag.nCols) {
memcpy(cfgRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);
}
// encode and send response
rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);
if (rspLen < 0) {
code = TSDB_CODE_INVALID_MSG;
goto _exit;
}
pRsp = rpcMallocCont(rspLen);
if (pRsp == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _exit;
}
tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp);
_exit:
rpcMsg.info = pMsg->info;
rpcMsg.pCont = pRsp;
rpcMsg.contLen = rspLen;
rpcMsg.code = code;
if (code) {
qError("get table %s cfg failed cause of %s", cfgReq.tbName, tstrerror(code));
}
tmsgSendRsp(&rpcMsg);
tFreeSTableCfgRsp(&cfgRsp);
metaReaderClear(&mer2);
metaReaderClear(&mer1);
return TSDB_CODE_SUCCESS;
}
int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
pLoad->vgId = TD_VID(pVnode);
pLoad->syncState = syncGetMyRole(pVnode->sync);
......@@ -157,4 +266,4 @@ tsdbReaderT tsdbQueryCacheLast(SVnode *pVnode, SQueryTableDataCond *pCond, STabl
return tsdbQueryCacheLastT(pVnode->pTsdb, pCond, groupList, qId, pMemRef);
#endif
return 0;
}
\ No newline at end of file
}
source/dnode/vnode/src/vnd/vnodeSvr.c
浏览文件 @ a23418d1
......@@ -255,6 +255,8 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);
case TDMT_VND_TABLE_META:
return vnodeGetTableMeta(pVnode, pMsg);
case TDMT_VND_TABLE_CFG:
return vnodeGetTableCfg(pVnode, pMsg);
case TDMT_VND_CONSUME:
return tqProcessPollReq(pVnode->pTq, pMsg, pInfo->workerId);
case TDMT_STREAM_TASK_RUN:
......
source/libs/catalog/inc/catalogInt.h
浏览文件 @ a23418d1
......@@ -71,11 +71,18 @@ typedef enum {
CTG_TASK_GET_TB_META,
CTG_TASK_GET_TB_HASH,
CTG_TASK_GET_TB_INDEX,
CTG_TASK_GET_TB_CFG,
CTG_TASK_GET_INDEX,
CTG_TASK_GET_UDF,
CTG_TASK_GET_USER,
} CTG_TASK_TYPE;
typedef enum {
CTG_TASK_LAUNCHED = 1,
CTG_TASK_DONE,
} CTG_TASK_STATUS;
typedef struct SCtgDebug {
bool lockEnable;
bool cacheEnable;
......@@ -102,6 +109,12 @@ typedef struct SCtgTbIndexCtx {
SName* pName;
} SCtgTbIndexCtx;
typedef struct SCtgTbCfgCtx {
SName* pName;
int32_t tbType;
SVgroupInfo* pVgInfo;
} SCtgTbCfgCtx;
typedef struct SCtgDbVgCtx {
char dbFName[TSDB_DB_FNAME_LEN];
} SCtgDbVgCtx;
......@@ -190,7 +203,9 @@ typedef struct SCtgJob {
SArray* pTasks;
int32_t taskDone;
SMetaData jobRes;
int32_t taskIdx;
SRWLatch taskLock;
uint64_t queryId;
SCatalog* pCtg;
SRequestConnInfo conn;
......@@ -206,6 +221,7 @@ typedef struct SCtgJob {
int32_t userNum;
int32_t dbInfoNum;
int32_t tbIndexNum;
int32_t tbCfgNum;
} SCtgJob;
typedef struct SCtgMsgCtx {
......@@ -215,24 +231,44 @@ typedef struct SCtgMsgCtx {
char* target;
} SCtgMsgCtx;
typedef struct SCtgTask SCtgTask;
typedef int32_t (*ctgSubTaskCbFp)(SCtgTask*);
typedef struct SCtgSubRes {
CTG_TASK_TYPE type;
int32_t code;
void* res;
ctgSubTaskCbFp fp;
} SCtgSubRes;
typedef struct SCtgTask {
CTG_TASK_TYPE type;
int32_t taskId;
SCtgJob* pJob;
void* taskCtx;
SCtgMsgCtx msgCtx;
int32_t code;
void* res;
CTG_TASK_TYPE type;
int32_t taskId;
SCtgJob* pJob;
void* taskCtx;
SCtgMsgCtx msgCtx;
int32_t code;
void* res;
CTG_TASK_STATUS status;
SRWLatch lock;
SArray* pParents;
SCtgSubRes subRes;
} SCtgTask;
typedef int32_t (*ctgInitTaskFp)(SCtgJob*, int32_t, void*);
typedef int32_t (*ctgLanchTaskFp)(SCtgTask*);
typedef int32_t (*ctgHandleTaskMsgRspFp)(SCtgTask*, int32_t, const SDataBuf *, int32_t);
typedef int32_t (*ctgDumpTaskResFp)(SCtgTask*);
typedef int32_t (*ctgCloneTaskResFp)(SCtgTask*, void**);
typedef int32_t (*ctgCompTaskFp)(SCtgTask*, void*, bool*);
typedef struct SCtgAsyncFps {
ctgLanchTaskFp launchFp;
ctgInitTaskFp initFp;
ctgLanchTaskFp launchFp;
ctgHandleTaskMsgRspFp handleRspFp;
ctgDumpTaskResFp dumpResFp;
ctgDumpTaskResFp dumpResFp;
ctgCompTaskFp compFp;
ctgCloneTaskResFp cloneFp;
} SCtgAsyncFps;
typedef struct SCtgApiStat {
......@@ -520,6 +556,8 @@ int32_t ctgDropTbIndexEnqueue(SCatalog* pCtg, SName* pName, bool syncOp);
int32_t ctgOpDropTbIndex(SCtgCacheOperation *operation);
int32_t ctgOpUpdateTbIndex(SCtgCacheOperation *operation);
int32_t ctgOpClearCache(SCtgCacheOperation *operation);
int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, char* dbFName, char *tableName, int32_t *tbType);
int32_t ctgGetTbHashVgroupFromCache(SCatalog *pCtg, const SName *pTableName, SVgroupInfo **pVgroup);
......@@ -535,10 +573,14 @@ int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const
int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);
int32_t ctgGetTbMetaFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableMetaOutput* out, SCtgTask* pTask);
int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTask* pTask);
int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableCfg **out, SCtgTask* pTask);
int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg **out, SCtgTask* pTask);
int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgJob** job, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int32_t* taskNum);
int32_t ctgLaunchJob(SCtgJob *pJob);
int32_t ctgMakeAsyncRes(SCtgJob *pJob);
int32_t ctgLaunchSubTask(SCtgTask *pTask, CTG_TASK_TYPE type, ctgSubTaskCbFp fp, void* param);
int32_t ctgGetTbCfgCb(SCtgTask *pTask);
int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst);
int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput);
......@@ -559,6 +601,7 @@ char * ctgTaskTypeStr(CTG_TASK_TYPE type);
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo *pMsgSendInfo, int32_t msgType, SCtgTask* pTask);
int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes);
void ctgFreeSTableIndex(void *info);
void ctgClearSubTaskRes(SCtgSubRes *pRes);
extern SCatalogMgmt gCtgMgmt;
......
source/libs/catalog/src/catalog.c
浏览文件 @ a23418d1
......@@ -22,36 +22,6 @@
SCatalogMgmt gCtgMgmt = {0};
int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq) {
int32_t code = 0;
STableMeta* tblMeta = NULL;
SCtgTbMetaCtx tbCtx = {0};
tbCtx.flag = CTG_FLAG_UNKNOWN_STB;
tbCtx.pName = pTableName;
CTG_ERR_JRET(ctgReadTbMetaFromCache(pCtg, &tbCtx, &tblMeta));
if (NULL == tblMeta) {
ctgDebug("table already not in cache, db:%s, tblName:%s", pTableName->dbname, pTableName->tname);
return TSDB_CODE_SUCCESS;
}
char dbFName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFName);
if (TSDB_SUPER_TABLE == tblMeta->tableType) {
CTG_ERR_JRET(ctgDropStbMetaEnqueue(pCtg, dbFName, tbCtx.tbInfo.dbId, pTableName->tname, tblMeta->suid, syncReq));
} else {
CTG_ERR_JRET(ctgDropTbMetaEnqueue(pCtg, dbFName, tbCtx.tbInfo.dbId, pTableName->tname, syncReq));
}
_return:
taosMemoryFreeClear(tblMeta);
CTG_RET(code);
}
int32_t ctgGetDBVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* dbFName, SCtgDBCache** dbCache, SDBVgInfo **pInfo) {
int32_t code = 0;
......@@ -212,29 +182,6 @@ _return:
CTG_RET(code);
}
int32_t ctgGetTbMetaFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta) {
if (CTG_IS_SYS_DBNAME(ctx->pName->dbname)) {
CTG_FLAG_SET_SYS_DB(ctx->flag);
}
CTG_ERR_RET(ctgReadTbMetaFromCache(pCtg, ctx, pTableMeta));
if (*pTableMeta) {
if (CTG_FLAG_MATCH_STB(ctx->flag, (*pTableMeta)->tableType) &&
((!CTG_FLAG_IS_FORCE_UPDATE(ctx->flag)) || (CTG_FLAG_IS_SYS_DB(ctx->flag)))) {
return TSDB_CODE_SUCCESS;
}
taosMemoryFreeClear(*pTableMeta);
}
if (CTG_FLAG_IS_UNKNOWN_STB(ctx->flag)) {
CTG_FLAG_SET_STB(ctx->flag, ctx->tbInfo.tbType);
}
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTbMeta(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta) {
int32_t code = 0;
STableMetaOutput *output = NULL;
......@@ -381,6 +328,23 @@ _return:
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTbType(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, int32_t *tbType) {
char dbFName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFName);
CTG_ERR_RET(ctgReadTbTypeFromCache(pCtg, dbFName, pTableName->tname, tbType));
if (*tbType > 0) {
return TSDB_CODE_SUCCESS;
}
STableMeta* pMeta = NULL;
CTG_ERR_RET(catalogGetTableMeta(pCtg, pConn, pTableName, &pMeta));
*tbType = pMeta->tableType;
taosMemoryFree(pMeta);
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTbIndex(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pRes) {
CTG_ERR_RET(ctgReadTbIndexFromCache(pCtg, pTableName, pRes));
if (*pRes) {
......@@ -419,6 +383,20 @@ _return:
CTG_RET(code);
}
int32_t ctgGetTbCfg(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, STableCfg** pCfg) {
int32_t tbType = 0;
CTG_ERR_RET(ctgGetTbType(pCtg, pConn, pTableName, &tbType));
if (TSDB_SUPER_TABLE == tbType) {
CTG_ERR_RET(ctgGetTableCfgFromMnode(pCtg, pConn, pTableName, pCfg, NULL));
} else {
SVgroupInfo vgroupInfo = {0};
CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pConn, pTableName, &vgroupInfo));
CTG_ERR_RET(ctgGetTableCfgFromVnode(pCtg, pConn, pTableName, &vgroupInfo, pCfg, NULL));
}
CTG_RET(TSDB_CODE_SUCCESS);
}
int32_t ctgGetTbDistVgInfo(SCatalog* pCtg, SRequestConnInfo *pConn, SName* pTableName, SArray** pVgList) {
STableMeta *tbMeta = NULL;
......@@ -1211,6 +1189,23 @@ _return:
CTG_API_LEAVE(code);
}
int32_t catalogRefreshGetTableCfg(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg** pCfg) {
CTG_API_ENTER();
if (NULL == pCtg || NULL == pConn || NULL == pTableName || NULL == pCfg) {
CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
}
int32_t code = 0;
CTG_ERR_JRET(catalogRemoveTableMeta(pCtg, (SName*)pTableName));
CTG_ERR_JRET(ctgGetTbCfg(pCtg, pConn, (SName*)pTableName, pCfg));
_return:
CTG_API_LEAVE(code);
}
int32_t catalogGetUdfInfo(SCatalog* pCtg, SRequestConnInfo *pConn, const char* funcName, SFuncInfo* pInfo) {
CTG_API_ENTER();
......
source/libs/catalog/src/ctgAsync.c
浏览文件 @ a23418d1
此差异已折叠。
source/libs/catalog/src/ctgCache.c
浏览文件 @ a23418d1
......@@ -326,6 +326,7 @@ _return:
int32_t ctgAcquireTbIndexFromCache(SCatalog* pCtg, char *dbFName, char* tbName, SCtgDBCache **pDb, SCtgTbCache** pTb) {
SCtgDBCache *dbCache = NULL;
SCtgTbCache* pCache = NULL;
ctgAcquireDBCache(pCtg, dbFName, &dbCache);
if (NULL == dbCache) {
ctgDebug("db %s not in cache", dbFName);
......@@ -333,7 +334,7 @@ int32_t ctgAcquireTbIndexFromCache(SCatalog* pCtg, char *dbFName, char* tbName,
}
int32_t sz = 0;
SCtgTbCache* pCache = taosHashAcquire(dbCache->tbCache, tbName, strlen(tbName));
pCache = taosHashAcquire(dbCache->tbCache, tbName, strlen(tbName));
if (NULL == pCache) {
ctgDebug("tb %s not in cache, dbFName:%s", tbName, dbFName);
goto _return;
......@@ -540,10 +541,10 @@ int32_t ctgReadTbVerFromCache(SCatalog *pCtg, SName *pTableName, int32_t *sver,
}
int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, char* dbFName, char *tableName, int32_t *tbType) {
int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, char* dbFName, char *tbName, int32_t *tbType) {
SCtgDBCache *dbCache = NULL;
SCtgTbCache *tbCache = NULL;
CTG_ERR_RET(ctgAcquireTbMetaFromCache(pCtg, dbFName, tableName, &dbCache, &tbCache));
CTG_ERR_RET(ctgAcquireTbMetaFromCache(pCtg, dbFName, tbName, &dbCache, &tbCache));
if (NULL == tbCache) {
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
return TSDB_CODE_SUCCESS;
......@@ -552,7 +553,7 @@ int32_t ctgReadTbTypeFromCache(SCatalog* pCtg, char* dbFName, char *tableName, i
*tbType = tbCache->pMeta->tableType;
ctgReleaseTbMetaToCache(pCtg, dbCache, tbCache);
ctgDebug("Got tb %s tbType %d from cache, dbFName:%s", tableName, *tbType, dbFName);
ctgDebug("Got tb %s tbType %d from cache, dbFName:%s", tbName, *tbType, dbFName);
return TSDB_CODE_SUCCESS;
}
......@@ -1394,7 +1395,7 @@ int32_t ctgWriteTbMetaToCache(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFNam
if (orig) {
origType = orig->tableType;
if (origType == meta->tableType && orig->uid == meta->uid && orig->sversion >= meta->sversion && orig->tversion >= meta->tversion) {
if (origType == meta->tableType && orig->uid == meta->uid && (origType == TSDB_CHILD_TABLE || (orig->sversion >= meta->sversion && orig->tversion >= meta->tversion))) {
taosMemoryFree(meta);
ctgDebug("ignore table %s meta update", tbName);
return TSDB_CODE_SUCCESS;
......@@ -2060,4 +2061,92 @@ int32_t ctgStartUpdateThread() {
}
int32_t ctgGetTbMetaFromCache(SCatalog* pCtg, SRequestConnInfo *pConn, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta) {
if (CTG_IS_SYS_DBNAME(ctx->pName->dbname)) {
CTG_FLAG_SET_SYS_DB(ctx->flag);
}
CTG_ERR_RET(ctgReadTbMetaFromCache(pCtg, ctx, pTableMeta));
if (*pTableMeta) {
if (CTG_FLAG_MATCH_STB(ctx->flag, (*pTableMeta)->tableType) &&
((!CTG_FLAG_IS_FORCE_UPDATE(ctx->flag)) || (CTG_FLAG_IS_SYS_DB(ctx->flag)))) {
return TSDB_CODE_SUCCESS;
}
taosMemoryFreeClear(*pTableMeta);
}
if (CTG_FLAG_IS_UNKNOWN_STB(ctx->flag)) {
CTG_FLAG_SET_STB(ctx->flag, ctx->tbInfo.tbType);
}
return TSDB_CODE_SUCCESS;
}
int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq) {
int32_t code = 0;
STableMeta* tblMeta = NULL;
SCtgTbMetaCtx tbCtx = {0};
tbCtx.flag = CTG_FLAG_UNKNOWN_STB;
tbCtx.pName = pTableName;
CTG_ERR_JRET(ctgReadTbMetaFromCache(pCtg, &tbCtx, &tblMeta));
if (NULL == tblMeta) {
ctgDebug("table already not in cache, db:%s, tblName:%s", pTableName->dbname, pTableName->tname);
return TSDB_CODE_SUCCESS;
}
char dbFName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFName);
if (TSDB_SUPER_TABLE == tblMeta->tableType) {
CTG_ERR_JRET(ctgDropStbMetaEnqueue(pCtg, dbFName, tbCtx.tbInfo.dbId, pTableName->tname, tblMeta->suid, syncReq));
} else {
CTG_ERR_JRET(ctgDropTbMetaEnqueue(pCtg, dbFName, tbCtx.tbInfo.dbId, pTableName->tname, syncReq));
}
_return:
taosMemoryFreeClear(tblMeta);
CTG_RET(code);
}
int32_t ctgGetTbHashVgroupFromCache(SCatalog *pCtg, const SName *pTableName, SVgroupInfo **pVgroup) {
if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
SCtgDBCache* dbCache = NULL;
int32_t code = 0;
char dbFName[TSDB_DB_FNAME_LEN] = {0};
tNameGetFullDbName(pTableName, dbFName);
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
if (NULL == dbCache) {
*pVgroup = NULL;
return TSDB_CODE_SUCCESS;
}
*pVgroup = taosMemoryCalloc(1, sizeof(SVgroupInfo));
CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, dbCache->vgCache.vgInfo, pTableName, *pVgroup));
_return:
if (dbCache) {
ctgReleaseVgInfoToCache(pCtg, dbCache);
}
if (code) {
taosMemoryFreeClear(*pVgroup);
}
CTG_RET(code);
}
source/libs/catalog/src/ctgRemote.c
浏览文件 @ a23418d1
......@@ -172,6 +172,39 @@ int32_t ctgProcessRspMsg(void* out, int32_t reqType, char* msg, int32_t msgSize,
qDebug("Got table meta from vnode, tbFName:%s", target);
break;
}
case TDMT_VND_TABLE_CFG: {
if (TSDB_CODE_SUCCESS != rspCode) {
qError("error rsp for table cfg from vnode, code:%s, tbFName:%s", tstrerror(rspCode), target);
CTG_ERR_RET(rspCode);
}
code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
if (code) {
qError("Process vnode tb cfg rsp failed, code:%s, tbFName:%s", tstrerror(code), target);
CTG_ERR_RET(code);
}
qDebug("Got table cfg from vnode, tbFName:%s", target);
break;
}
case TDMT_MND_TABLE_CFG: {
if (TSDB_CODE_SUCCESS != rspCode) {
qError("error rsp for stb cfg from mnode, error:%s, tbFName:%s", tstrerror(rspCode), target);
CTG_ERR_RET(rspCode);
}
code = queryProcessMsgRsp[TMSG_INDEX(reqType)](out, msg, msgSize);
if (code) {
qError("Process mnode stb cfg rsp failed, error:%s, tbFName:%s", tstrerror(code), target);
CTG_ERR_RET(code);
}
qDebug("Got stb cfg from mnode, tbFName:%s", target);
break;
}
default:
qError("invalid req type %s", TMSG_INFO(reqType));
return TSDB_CODE_APP_ERROR;
}
return TSDB_CODE_SUCCESS;
......@@ -550,7 +583,7 @@ int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const
int32_t ctgGetTbMetaFromMnodeImpl(SCatalog* pCtg, SRequestConnInfo *pConn, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask) {
SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
SBuildTableInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
char *msg = NULL;
SEpSet *pVnodeEpSet = NULL;
int32_t msgLen = 0;
......@@ -606,9 +639,11 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SNa
sprintf(tbFName, "%s.%s", dbFName, pTableName->tname);
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
ctgDebug("try to get table meta from vnode, vgId:%d, tbFName:%s", vgroupInfo->vgId, tbFName);
SEp* pEp = &vgroupInfo->epSet.eps[vgroupInfo->epSet.inUse];
ctgDebug("try to get table meta from vnode, vgId:%d, ep num:%d, ep %s:%d, tbFName:%s",
vgroupInfo->vgId, vgroupInfo->epSet.numOfEps, pEp->fqdn, pEp->port, tbFName);
SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
SBuildTableInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
char *msg = NULL;
int32_t msgLen = 0;
......@@ -646,4 +681,89 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SNa
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, SVgroupInfo *vgroupInfo, STableCfg **out, SCtgTask* pTask) {
char *msg = NULL;
int32_t msgLen = 0;
int32_t reqType = TDMT_VND_TABLE_CFG;
char tbFName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pTableName, tbFName);
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
char dbFName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFName);
SBuildTableInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char*)pTableName->tname};
SEp* pEp = &vgroupInfo->epSet.eps[vgroupInfo->epSet.inUse];
ctgDebug("try to get table cfg from vnode, vgId:%d, ep num:%d, ep %s:%d, tbFName:%s",
vgroupInfo->vgId, vgroupInfo->epSet.numOfEps, pEp->fqdn, pEp->port, tbFName);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen, mallocFp);
if (code) {
ctgError("Build get tb cfg msg failed, code:%s, tbFName:%s", tstrerror(code), tbFName);
CTG_ERR_RET(code);
}
if (pTask) {
CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, NULL, (char*)tbFName));
SRequestConnInfo vConn = {.pTrans = pConn->pTrans,
.requestId = pConn->requestId,
.requestObjRefId = pConn->requestObjRefId,
.mgmtEps = vgroupInfo->epSet};
CTG_RET(ctgAsyncSendMsg(pCtg, &vConn, pTask, reqType, msg, msgLen));
}
SRpcMsg rpcMsg = {
.msgType = reqType,
.pCont = msg,
.contLen = msgLen,
};
SRpcMsg rpcRsp = {0};
rpcSendRecv(pConn->pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
return TSDB_CODE_SUCCESS;
}
int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg **out, SCtgTask* pTask) {
char *msg = NULL;
int32_t msgLen = 0;
int32_t reqType = TDMT_MND_TABLE_CFG;
char tbFName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pTableName, tbFName);
void*(*mallocFp)(int32_t) = pTask ? taosMemoryMalloc : rpcMallocCont;
char dbFName[TSDB_DB_FNAME_LEN];
tNameGetFullDbName(pTableName, dbFName);
SBuildTableInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = (char*)pTableName->tname};
ctgDebug("try to get table cfg from mnode, tbFName:%s", tbFName);
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen, mallocFp);
if (code) {
ctgError("Build get tb cfg msg failed, code:%s, tbFName:%s", tstrerror(code), tbFName);
CTG_ERR_RET(code);
}
if (pTask) {
CTG_ERR_RET(ctgUpdateMsgCtx(&pTask->msgCtx, reqType, NULL, (char*)tbFName));
CTG_RET(ctgAsyncSendMsg(pCtg, pConn, pTask, reqType, msg, msgLen));
}
SRpcMsg rpcMsg = {
.msgType = reqType,
.pCont = msg,
.contLen = msgLen,
};
SRpcMsg rpcRsp = {0};
rpcSendRecv(pConn->pTrans, &pConn->mgmtEps, &rpcMsg, &rpcRsp);
CTG_ERR_RET(ctgProcessRspMsg(out, reqType, rpcRsp.pCont, rpcRsp.contLen, rpcRsp.code, (char*)tbFName));
return TSDB_CODE_SUCCESS;
}
source/libs/catalog/src/ctgUtil.c
浏览文件 @ a23418d1
......@@ -33,6 +33,10 @@ char *ctgTaskTypeStr(CTG_TASK_TYPE type) {
return "[get table meta]";
case CTG_TASK_GET_TB_HASH:
return "[get table hash]";
case CTG_TASK_GET_TB_INDEX:
return "[get table index]";
case CTG_TASK_GET_TB_CFG:
return "[get table cfg]";
case CTG_TASK_GET_INDEX:
return "[get index]";
case CTG_TASK_GET_UDF:
......@@ -96,6 +100,9 @@ void ctgFreeSMetaData(SMetaData* pData) {
taosArrayDestroy(pData->pQnodeList);
pData->pQnodeList = NULL;
taosArrayDestroy(pData->pTableCfg);
pData->pTableCfg = NULL;
}
void ctgFreeSCtgUserAuth(SCtgUserAuth *userCache) {
......@@ -280,6 +287,13 @@ void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) {
}
break;
}
case TDMT_VND_TABLE_CFG:
case TDMT_MND_TABLE_CFG: {
STableCfgRsp* pOut = (STableCfgRsp*)pCtx->out;
tFreeSTableCfgRsp(pOut);
taosMemoryFreeClear(pCtx->out);
break;
}
case TDMT_MND_RETRIEVE_FUNC: {
SFuncInfo* pOut = (SFuncInfo*)pCtx->out;
taosMemoryFree(pOut->pCode);
......@@ -328,14 +342,151 @@ void ctgResetTbMetaTask(SCtgTask* pTask) {
taosMemoryFreeClear(pTask->res);
}
void ctgFreeTask(SCtgTask* pTask) {
ctgFreeMsgCtx(&pTask->msgCtx);
void ctgFreeTaskRes(CTG_TASK_TYPE type, void **pRes) {
switch (type) {
case CTG_TASK_GET_QNODE: {
taosArrayDestroy((SArray*)*pRes);
*pRes = NULL;
break;
}
case CTG_TASK_GET_TB_META: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_DB_VGROUP: {
taosArrayDestroy((SArray*)*pRes);
*pRes = NULL;
break;
}
case CTG_TASK_GET_DB_CFG: {
if (*pRes) {
SDbCfgInfo* pInfo = (SDbCfgInfo*)*pRes;
taosArrayDestroy(pInfo->pRetensions);
taosMemoryFreeClear(*pRes);
}
break;
}
case CTG_TASK_GET_DB_INFO: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_TB_HASH: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_TB_INDEX: {
taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
*pRes = NULL;
break;
}
case CTG_TASK_GET_TB_CFG: {
if (*pRes) {
STableCfg* pInfo = (STableCfg*)*pRes;
tFreeSTableCfgRsp(pInfo);
taosMemoryFreeClear(*pRes);
}
break;
}
case CTG_TASK_GET_INDEX: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_UDF: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_USER: {
taosMemoryFreeClear(*pRes);
break;
}
default:
qError("invalid task type %d", type);
break;
}
}
void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void **pRes) {
switch (type) {
case CTG_TASK_GET_QNODE: {
taosArrayDestroy((SArray*)*pRes);
*pRes = NULL;
break;
}
case CTG_TASK_GET_TB_META: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_DB_VGROUP: {
if (*pRes) {
SDBVgInfo* pInfo = (SDBVgInfo*)*pRes;
taosHashCleanup(pInfo->vgHash);
taosMemoryFreeClear(*pRes);
}
break;
}
case CTG_TASK_GET_DB_CFG: {
if (*pRes) {
SDbCfgInfo* pInfo = (SDbCfgInfo*)*pRes;
taosArrayDestroy(pInfo->pRetensions);
taosMemoryFreeClear(*pRes);
}
break;
}
case CTG_TASK_GET_DB_INFO: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_TB_HASH: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_TB_INDEX: {
taosArrayDestroyEx(*pRes, tFreeSTableIndexInfo);
*pRes = NULL;
break;
}
case CTG_TASK_GET_TB_CFG: {
if (*pRes) {
STableCfg* pInfo = (STableCfg*)*pRes;
tFreeSTableCfgRsp(pInfo);
taosMemoryFreeClear(*pRes);
}
break;
}
case CTG_TASK_GET_INDEX: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_UDF: {
taosMemoryFreeClear(*pRes);
break;
}
case CTG_TASK_GET_USER: {
taosMemoryFreeClear(*pRes);
break;
}
default:
qError("invalid task type %d", type);
break;
}
}
void ctgClearSubTaskRes(SCtgSubRes *pRes) {
pRes->code = 0;
if (NULL == pRes->res) {
return;
}
ctgFreeSubTaskRes(pRes->type, &pRes->res);
}
void ctgFreeTaskCtx(SCtgTask* pTask) {
switch (pTask->type) {
case CTG_TASK_GET_QNODE: {
taosArrayDestroy((SArray*)pTask->res);
taosMemoryFreeClear(pTask->taskCtx);
pTask->res = NULL;
break;
}
case CTG_TASK_GET_TB_META: {
......@@ -346,56 +497,49 @@ void ctgFreeTask(SCtgTask* pTask) {
pTask->msgCtx.lastOut = NULL;
}
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res);
break;
}
case CTG_TASK_GET_DB_VGROUP: {
taosArrayDestroy((SArray*)pTask->res);
taosMemoryFreeClear(pTask->taskCtx);
pTask->res = NULL;
break;
}
case CTG_TASK_GET_DB_CFG: {
taosMemoryFreeClear(pTask->taskCtx);
if (pTask->res) {
SDbCfgInfo* pInfo = (SDbCfgInfo*)pTask->res;
taosArrayDestroy(pInfo->pRetensions);
taosMemoryFreeClear(pTask->res);
}
break;
}
case CTG_TASK_GET_DB_INFO: {
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res);
break;
}
case CTG_TASK_GET_TB_HASH: {
SCtgTbHashCtx* taskCtx = (SCtgTbHashCtx*)pTask->taskCtx;
taosMemoryFreeClear(taskCtx->pName);
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res);
break;
}
case CTG_TASK_GET_TB_INDEX: {
SCtgTbIndexCtx* taskCtx = (SCtgTbIndexCtx*)pTask->taskCtx;
taosMemoryFreeClear(taskCtx->pName);
taosMemoryFreeClear(pTask->taskCtx);
taosArrayDestroyEx(pTask->res, tFreeSTableIndexInfo);
break;
}
case CTG_TASK_GET_TB_CFG: {
SCtgTbCfgCtx* taskCtx = (SCtgTbCfgCtx*)pTask->taskCtx;
taosMemoryFreeClear(taskCtx->pName);
taosMemoryFreeClear(taskCtx->pVgInfo);
taosMemoryFreeClear(pTask->taskCtx);
break;
}
case CTG_TASK_GET_INDEX: {
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res);
break;
}
case CTG_TASK_GET_UDF: {
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res);
break;
}
case CTG_TASK_GET_USER: {
taosMemoryFreeClear(pTask->taskCtx);
taosMemoryFreeClear(pTask->res);
break;
}
default:
......@@ -404,6 +548,16 @@ void ctgFreeTask(SCtgTask* pTask) {
}
}
void ctgFreeTask(SCtgTask* pTask) {
ctgFreeMsgCtx(&pTask->msgCtx);
ctgFreeTaskRes(pTask->type, &pTask->res);
ctgFreeTaskCtx(pTask);
taosArrayDestroy(pTask->pParents);
ctgClearSubTaskRes(&pTask->subRes);
}
void ctgFreeTasks(SArray* pArray) {
if (NULL == pArray) {
return;
......
source/libs/command/src/command.c
浏览文件 @ a23418d1
......@@ -16,6 +16,7 @@
#include "command.h"
#include "catalog.h"
#include "tdatablock.h"
#include "tglobal.h"
static int32_t getSchemaBytes(const SSchema* pSchema) {
switch (pSchema->type) {
......@@ -123,11 +124,430 @@ static int32_t execDescribe(SNode* pStmt, SRetrieveTableRsp** pRsp) {
static int32_t execResetQueryCache() { return catalogClearCache(); }
static int32_t execShowCreateDatabase(SShowCreateDatabaseStmt* pStmt) { return TSDB_CODE_FAILED; }
static int32_t execShowCreateTable(SShowCreateTableStmt* pStmt) { return TSDB_CODE_FAILED; }
static SSDataBlock* buildCreateDBResultDataBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = SHOW_CREATE_DB_RESULT_COLS;
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_DB_RESULT_FIELD1_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_DB_RESULT_FIELD2_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
return pBlock;
}
int64_t getValOfDiffPrecision(int8_t unit, int64_t val) {
int64_t v = 0;
switch (unit) {
case 's':
v = val / 1000;
break;
case 'm':
v = val / tsTickPerMin[TSDB_TIME_PRECISION_MILLI];
break;
case 'h':
v = val / (tsTickPerMin[TSDB_TIME_PRECISION_MILLI] * 60);
break;
case 'd':
v = val / (tsTickPerMin[TSDB_TIME_PRECISION_MILLI] * 24 * 60);
break;
case 'w':
v = val / (tsTickPerMin[TSDB_TIME_PRECISION_MILLI] * 24 * 60 * 7);
break;
default:
break;
}
return v;
}
char *buildRetension(SArray *pRetension) {
size_t size = taosArrayGetSize(pRetension);
if (size == 0) {
return NULL;
}
char *p1 = taosMemoryCalloc(1, 100);
SRetention *p = taosArrayGet(pRetension, 0);
int32_t len = 0;
int64_t v1 = getValOfDiffPrecision(p->freqUnit, p->freq);
int64_t v2 = getValOfDiffPrecision(p->keepUnit, p->keep);
len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit);
if (size > 1) {
len += sprintf(p1 + len, ",");
p = taosArrayGet(pRetension, 1);
v1 = getValOfDiffPrecision(p->freqUnit, p->freq);
v2 = getValOfDiffPrecision(p->keepUnit, p->keep);
len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit);
}
if (size > 2) {
len += sprintf(p1 + len, ",");
p = taosArrayGet(pRetension, 2);
v1 = getValOfDiffPrecision(p->freqUnit, p->freq);
v2 = getValOfDiffPrecision(p->keepUnit, p->keep);
len += sprintf(p1 + len, "%" PRId64 "%c:%" PRId64 "%c", v1, p->freqUnit, v2, p->keepUnit);
}
return p1;
}
static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char *dbFName, SDbCfgInfo* pCfg) {
blockDataEnsureCapacity(pBlock, 1);
pBlock->info.rows = 1;
SColumnInfoData* pCol1 = taosArrayGet(pBlock->pDataBlock, 0);
char buf1[SHOW_CREATE_DB_RESULT_FIELD1_LEN] = {0};
STR_TO_VARSTR(buf1, dbFName);
colDataAppend(pCol1, 0, buf1, false);
SColumnInfoData* pCol2 = taosArrayGet(pBlock->pDataBlock, 1);
char buf2[SHOW_CREATE_DB_RESULT_FIELD2_LEN] = {0};
int32_t len = 0;
char *prec = NULL;
switch (pCfg->precision) {
case TSDB_TIME_PRECISION_MILLI:
prec = TSDB_TIME_PRECISION_MILLI_STR;
break;
case TSDB_TIME_PRECISION_MICRO:
prec = TSDB_TIME_PRECISION_MICRO_STR;
break;
case TSDB_TIME_PRECISION_NANO:
prec = TSDB_TIME_PRECISION_NANO_STR;
break;
default:
prec = "none";
break;
}
char *retentions = buildRetension(pCfg->pRetensions);
len += sprintf(buf2 + VARSTR_HEADER_SIZE, "CREATE DATABASE `%s` BUFFER %d CACHELAST %d COMP %d DURATION %dm "
"FSYNC %d MAXROWS %d MINROWS %d KEEP %dm,%dm,%dm PAGES %d PAGESIZE %d PRECISION '%s' REPLICA %d "
"STRICT %d WAL %d VGROUPS %d SINGLE_STABLE %d",
dbFName, pCfg->buffer, pCfg->cacheLastRow, pCfg->compression, pCfg->daysPerFile,
pCfg->fsyncPeriod, pCfg->maxRows, pCfg->minRows, pCfg->daysToKeep0, pCfg->daysToKeep1, pCfg->daysToKeep2,
pCfg->pages, pCfg->pageSize, prec, pCfg->replications, pCfg->strict, pCfg->walLevel, pCfg->numOfVgroups,
1 == pCfg->numOfStables);
if (retentions) {
len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, " RETENTIONS %s", retentions);
taosMemoryFree(retentions);
}
(varDataLen(buf2)) = len;
colDataAppend(pCol2, 0, buf2, false);
}
static int32_t execShowCreateDatabase(SShowCreateDatabaseStmt* pStmt, SRetrieveTableRsp** pRsp) {
SSDataBlock* pBlock = buildCreateDBResultDataBlock();
setCreateDBResultIntoDataBlock(pBlock, pStmt->dbName, pStmt->pCfg);
size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
*pRsp = taosMemoryCalloc(1, rspSize);
if (NULL == *pRsp) {
return TSDB_CODE_OUT_OF_MEMORY;
}
(*pRsp)->useconds = 0;
(*pRsp)->completed = 1;
(*pRsp)->precision = 0;
(*pRsp)->compressed = 0;
(*pRsp)->compLen = 0;
(*pRsp)->numOfRows = htonl(1);
(*pRsp)->numOfCols = htonl(SHOW_CREATE_DB_RESULT_COLS);
int32_t len = 0;
blockCompressEncode(pBlock, (*pRsp)->data, &len, SHOW_CREATE_DB_RESULT_COLS, false);
ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
blockDataDestroy(pBlock);
return TSDB_CODE_SUCCESS;
}
static SSDataBlock* buildCreateTbResultDataBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = SHOW_CREATE_TB_RESULT_COLS;
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
static int32_t execShowCreateSTable(SShowCreateTableStmt* pStmt) { return TSDB_CODE_FAILED; }
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_TB_RESULT_FIELD1_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_TB_RESULT_FIELD2_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
return pBlock;
}
void appendColumnFields(char* buf, int32_t* len, STableCfg* pCfg) {
for (int32_t i = 0; i < pCfg->numOfColumns; ++i) {
SSchema* pSchema = pCfg->pSchemas + i;
char type[32];
sprintf(type, "%s", tDataTypes[pSchema->type].name);
if (TSDB_DATA_TYPE_VARCHAR == pSchema->type) {
sprintf(type + strlen(type), "(%d)", (int32_t)(pSchema->bytes - VARSTR_HEADER_SIZE));
} else if (TSDB_DATA_TYPE_NCHAR == pSchema->type) {
sprintf(type + strlen(type), "(%d)", (int32_t)((pSchema->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE));
}
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s`%s` %s", ((i > 0) ? ", " : ""), pSchema->name, type);
}
}
void appendTagFields(char* buf, int32_t* len, STableCfg* pCfg) {
for (int32_t i = 0; i < pCfg->numOfTags; ++i) {
SSchema* pSchema = pCfg->pSchemas + pCfg->numOfColumns + i;
char type[32];
sprintf(type, "%s", tDataTypes[pSchema->type].name);
if (TSDB_DATA_TYPE_VARCHAR == pSchema->type) {
sprintf(type + strlen(type), "(%d)", (int32_t)(pSchema->bytes - VARSTR_HEADER_SIZE));
} else if (TSDB_DATA_TYPE_NCHAR == pSchema->type) {
sprintf(type + strlen(type), "(%d)", (int32_t)((pSchema->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE));
}
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s`%s` %s", ((i > 0) ? ", " : ""), pSchema->name, type);
}
}
void appendTagNameFields(char* buf, int32_t* len, STableCfg* pCfg) {
for (int32_t i = 0; i < pCfg->numOfTags; ++i) {
SSchema* pSchema = pCfg->pSchemas + pCfg->numOfColumns + i;
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s`%s`", ((i > 0) ? ", " : ""), pSchema->name);
}
}
int32_t appendTagValues(char* buf, int32_t* len, STableCfg* pCfg) {
SArray *pTagVals = NULL;
STag *pTag = (STag*)pCfg->pTags;
if (pCfg->pTags && pTag->flags & TD_TAG_JSON) {
char *pJson = parseTagDatatoJson(pTag);
if (pJson) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s", pJson);
taosMemoryFree(pJson);
}
return TSDB_CODE_SUCCESS;
}
int32_t code = tTagToValArray((const STag *)pCfg->pTags, &pTagVals);
if (code) {
return code;
}
int16_t valueNum = taosArrayGetSize(pTagVals);
int32_t num = 0;
int32_t j = 0;
for (int32_t i = 0; i < pCfg->numOfTags; ++i) {
SSchema* pSchema = pCfg->pSchemas + pCfg->numOfColumns + i;
if (i > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ", ");
}
if (j >= valueNum) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "NULL");
continue;
}
STagVal *pTagVal = (STagVal *)taosArrayGet(pTagVals, j);
if (pSchema->colId > pTagVal->cid) {
qError("tag value and column mismatch, schemaId:%d, valId:%d", pSchema->colId, pTagVal->cid);
taosArrayDestroy(pTagVals);
return TSDB_CODE_APP_ERROR;
} else if (pSchema->colId == pTagVal->cid) {
char type = pTagVal->type;
int32_t tlen = 0;
if (IS_VAR_DATA_TYPE(type)) {
dataConverToStr(buf + VARSTR_HEADER_SIZE + *len, type, pTagVal->pData, pTagVal->nData, &tlen);
} else {
dataConverToStr(buf + VARSTR_HEADER_SIZE + *len, type, &pTagVal->i64, tDataTypes[type].bytes, &tlen);
}
*len += tlen;
j++;
} else {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "NULL");
}
/*
if (type == TSDB_DATA_TYPE_BINARY) {
if (pTagVal->nData > 0) {
if (num) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ", ");
}
memcpy(buf + VARSTR_HEADER_SIZE + *len, pTagVal->pData, pTagVal->nData);
*len += pTagVal->nData;
}
} else if (type == TSDB_DATA_TYPE_NCHAR) {
if (pTagVal->nData > 0) {
if (num) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ", ");
}
int32_t tlen = taosUcs4ToMbs((TdUcs4 *)pTagVal->pData, pTagVal->nData, buf + VARSTR_HEADER_SIZE + *len);
}
} else if (type == TSDB_DATA_TYPE_DOUBLE) {
double val = *(double *)(&pTagVal->i64);
int len = 0;
term = indexTermCreate(suid, ADD_VALUE, type, key, nKey, (const char *)&val, len);
} else if (type == TSDB_DATA_TYPE_BOOL) {
int val = *(int *)(&pTagVal->i64);
int len = 0;
term = indexTermCreate(suid, ADD_VALUE, TSDB_DATA_TYPE_INT, key, nKey, (const char *)&val, len);
}
*/
}
taosArrayDestroy(pTagVals);
return TSDB_CODE_SUCCESS;
}
void appendTableOptions(char* buf, int32_t* len, STableCfg* pCfg) {
if (pCfg->commentLen > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " COMMENT '%s'", pCfg->pComment);
} else if (0 == pCfg->commentLen) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " COMMENT ''");
}
if (pCfg->watermark1 > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " WATERMARK %" PRId64 "a", pCfg->watermark1);
if (pCfg->watermark2 > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ", %" PRId64 "a", pCfg->watermark2);
}
}
if (pCfg->delay1 > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " MAX_DELAY %" PRId64 "a", pCfg->delay1);
if (pCfg->delay2 > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ", %" PRId64 "a", pCfg->delay2);
}
}
int32_t funcNum = taosArrayGetSize(pCfg->pFuncs);
if (funcNum > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " ROLLUP(");
for (int32_t i = 0; i < funcNum; ++i) {
char* pFunc = taosArrayGet(pCfg->pFuncs, i);
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, "%s%s", ((i > 0) ? ", " : ""), pFunc);
}
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, ")");
}
if (pCfg->ttl > 0) {
*len += sprintf(buf + VARSTR_HEADER_SIZE + *len, " TTL %d", pCfg->ttl);
}
}
static int32_t setCreateTBResultIntoDataBlock(SSDataBlock* pBlock, char *tbName, STableCfg* pCfg) {
int32_t code = 0;
blockDataEnsureCapacity(pBlock, 1);
pBlock->info.rows = 1;
SColumnInfoData* pCol1 = taosArrayGet(pBlock->pDataBlock, 0);
char buf1[SHOW_CREATE_TB_RESULT_FIELD1_LEN] = {0};
STR_TO_VARSTR(buf1, tbName);
colDataAppend(pCol1, 0, buf1, false);
SColumnInfoData* pCol2 = taosArrayGet(pBlock->pDataBlock, 1);
char buf2[SHOW_CREATE_TB_RESULT_FIELD2_LEN] = {0};
int32_t len = 0;
if (TSDB_SUPER_TABLE == pCfg->tableType) {
len += sprintf(buf2 + VARSTR_HEADER_SIZE, "CREATE STABLE `%s` (", tbName);
appendColumnFields(buf2, &len, pCfg);
len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, ") TAGS (");
appendTagFields(buf2, &len, pCfg);
len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, ")");
appendTableOptions(buf2, &len, pCfg);
} else if (TSDB_CHILD_TABLE == pCfg->tableType) {
len += sprintf(buf2 + VARSTR_HEADER_SIZE, "CREATE TABLE `%s` USING `%s` (", tbName, pCfg->stbName);
appendTagNameFields(buf2, &len, pCfg);
len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, ") TAGS (");
code = appendTagValues(buf2, &len, pCfg);
if (code) {
return code;
}
len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, ")");
appendTableOptions(buf2, &len, pCfg);
} else {
len += sprintf(buf2 + VARSTR_HEADER_SIZE, "CREATE TABLE `%s` (", tbName);
appendColumnFields(buf2, &len, pCfg);
len += sprintf(buf2 + VARSTR_HEADER_SIZE + len, ")");
}
varDataLen(buf2) = len;
colDataAppend(pCol2, 0, buf2, false);
return TSDB_CODE_SUCCESS;
}
static int32_t execShowCreateTable(SShowCreateTableStmt* pStmt, SRetrieveTableRsp** pRsp) {
SSDataBlock* pBlock = buildCreateTbResultDataBlock();
int32_t code = setCreateTBResultIntoDataBlock(pBlock, pStmt->tableName, pStmt->pCfg);
if (code) {
return code;
}
size_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
*pRsp = taosMemoryCalloc(1, rspSize);
if (NULL == *pRsp) {
return TSDB_CODE_OUT_OF_MEMORY;
}
(*pRsp)->useconds = 0;
(*pRsp)->completed = 1;
(*pRsp)->precision = 0;
(*pRsp)->compressed = 0;
(*pRsp)->compLen = 0;
(*pRsp)->numOfRows = htonl(1);
(*pRsp)->numOfCols = htonl(SHOW_CREATE_TB_RESULT_COLS);
int32_t len = 0;
blockCompressEncode(pBlock, (*pRsp)->data, &len, SHOW_CREATE_TB_RESULT_COLS, false);
ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
blockDataDestroy(pBlock);
return TSDB_CODE_SUCCESS;
}
static int32_t execShowCreateSTable(SShowCreateTableStmt* pStmt, SRetrieveTableRsp** pRsp) {
STableCfg* pCfg = (STableCfg*)pStmt->pCfg;
if (TSDB_SUPER_TABLE != pCfg->tableType) {
terrno = TSDB_CODE_TSC_NOT_STABLE_ERROR;
return terrno;
}
return execShowCreateTable(pStmt, pRsp);
}
static int32_t execAlterLocal(SAlterLocalStmt* pStmt) { return TSDB_CODE_FAILED; }
......@@ -140,11 +560,11 @@ int32_t qExecCommand(SNode* pStmt, SRetrieveTableRsp** pRsp) {
case QUERY_NODE_RESET_QUERY_CACHE_STMT:
return execResetQueryCache();
case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
return execShowCreateDatabase((SShowCreateDatabaseStmt*)pStmt);
return execShowCreateDatabase((SShowCreateDatabaseStmt*)pStmt, pRsp);
case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
return execShowCreateTable((SShowCreateTableStmt*)pStmt);
return execShowCreateTable((SShowCreateTableStmt*)pStmt, pRsp);
case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
return execShowCreateSTable((SShowCreateTableStmt*)pStmt);
return execShowCreateSTable((SShowCreateTableStmt*)pStmt, pRsp);
case QUERY_NODE_ALTER_LOCAL_STMT:
return execAlterLocal((SAlterLocalStmt*)pStmt);
case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:
......
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @ a23418d1
......@@ -655,7 +655,7 @@ void nodesDestroyNode(SNode* pNode) {
break;
case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
taosMemoryFreeClear(((SShowCreateTableStmt*)pNode)->pMeta);
taosMemoryFreeClear(((SShowCreateTableStmt*)pNode)->pCfg);
break;
case QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT: // no pointer field
case QUERY_NODE_KILL_CONNECTION_STMT: // no pointer field
......
source/libs/parser/inc/parUtil.h
浏览文件 @ a23418d1
......@@ -47,6 +47,7 @@ typedef struct SParseMetaCache {
SHashObj* pUserAuth; // key is SUserAuthInfo serialized string, element is bool indicating whether or not to pass
SHashObj* pUdf; // key is funcName, element is SFuncInfo*
SHashObj* pTableIndex; // key is tbFName, element is SArray<STableIndexInfo>*
SHashObj* pTableCfg; // key is tbFName, element is STableCfg*
SArray* pDnodes; // element is SEpSet
bool dnodeRequired;
} SParseMetaCache;
......@@ -79,6 +80,7 @@ int32_t reserveUserAuthInCache(int32_t acctId, const char* pUser, const char* pD
int32_t reserveUserAuthInCacheExt(const char* pUser, const SName* pName, AUTH_TYPE type, SParseMetaCache* pMetaCache);
int32_t reserveUdfInCache(const char* pFunc, SParseMetaCache* pMetaCache);
int32_t reserveTableIndexInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveTableCfgInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveDnodeRequiredInCache(SParseMetaCache* pMetaCache);
int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta);
int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo);
......@@ -90,6 +92,7 @@ int32_t getUserAuthFromCache(SParseMetaCache* pMetaCache, const char* pUser, con
bool* pPass);
int32_t getUdfInfoFromCache(SParseMetaCache* pMetaCache, const char* pFunc, SFuncInfo* pInfo);
int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName, SArray** pIndexes);
int32_t getTableCfgFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableCfg** pOutput);
int32_t getDnodeListFromCache(SParseMetaCache* pMetaCache, SArray** pDnodes);
void destoryParseMetaCache(SParseMetaCache* pMetaCache);
......
source/libs/parser/src/parAstParser.c
浏览文件 @ a23418d1
......@@ -411,7 +411,14 @@ static int32_t collectMetaKeyFromShowCreateDatabase(SCollectMetaKeyCxt* pCxt, SS
}
static int32_t collectMetaKeyFromShowCreateTable(SCollectMetaKeyCxt* pCxt, SShowCreateTableStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
SName name = {.type = TSDB_TABLE_NAME_T, .acctId = pCxt->pParseCxt->acctId};
strcpy(name.dbname, pStmt->dbName);
strcpy(name.tname, pStmt->tableName);
int32_t code = catalogRemoveTableMeta(pCxt->pParseCxt->pCatalog, &name);
if (TSDB_CODE_SUCCESS == code) {
code = reserveTableCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
}
return code;
}
static int32_t collectMetaKeyFromShowApps(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
......
source/libs/parser/src/parInsert.c
浏览文件 @ a23418d1
......@@ -1081,15 +1081,6 @@ end:
return code;
}
static int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
*pDst = taosMemoryMalloc(TABLE_META_SIZE(pSrc));
if (NULL == *pDst) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memcpy(*pDst, pSrc, TABLE_META_SIZE(pSrc));
return TSDB_CODE_SUCCESS;
}
static int32_t storeTableMeta(SInsertParseContext* pCxt, SHashObj* pHash, SName* pTableName, const char* pName,
int32_t len, STableMeta* pMeta) {
SVgroupInfo vg;
......
source/libs/parser/src/parTranslater.c
浏览文件 @ a23418d1
......@@ -133,6 +133,30 @@ static int32_t getTableMeta(STranslateContext* pCxt, const char* pDbName, const
return getTableMetaImpl(pCxt, toName(pCxt->pParseCxt->acctId, pDbName, pTableName, &name), pMeta);
}
static int32_t getTableCfg(STranslateContext* pCxt, const SName* pName, STableCfg** pCfg) {
SParseContext* pParCxt = pCxt->pParseCxt;
int32_t code = collectUseDatabase(pName, pCxt->pDbs);
if (TSDB_CODE_SUCCESS == code) {
code = collectUseTable(pName, pCxt->pTables);
}
if (TSDB_CODE_SUCCESS == code) {
if (pParCxt->async) {
code = getTableCfgFromCache(pCxt->pMetaCache, pName, pCfg);
} else {
SRequestConnInfo conn = {.pTrans = pParCxt->pTransporter,
.requestId = pParCxt->requestId,
.requestObjRefId = pParCxt->requestRid,
.mgmtEps = pParCxt->mgmtEpSet};
code = catalogRefreshGetTableCfg(pParCxt->pCatalog, &conn, pName, pCfg);
}
}
if (TSDB_CODE_SUCCESS != code) {
parserError("catalogRefreshGetTableCfg error, code:%s, dbName:%s, tbName:%s", tstrerror(code), pName->dbname,
pName->tname);
}
return code;
}
static int32_t refreshGetTableMeta(STranslateContext* pCxt, const char* pDbName, const char* pTableName,
STableMeta** pMeta) {
SParseContext* pParCxt = pCxt->pParseCxt;
......@@ -3503,13 +3527,24 @@ static int32_t buildRollupAst(STranslateContext* pCxt, SCreateTableStmt* pStmt,
return code;
}
static int32_t buildRollupFuncs(SNodeList* pFuncs, SArray** pArray) {
if (NULL == pFuncs) {
return TSDB_CODE_SUCCESS;
}
*pArray = taosArrayInit(LIST_LENGTH(pFuncs), TSDB_FUNC_NAME_LEN);
SNode* pNode;
FOREACH(pNode, pFuncs) {
taosArrayPush(*pArray, ((SFunctionNode*)pNode)->functionName);
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildCreateStbReq(STranslateContext* pCxt, SCreateTableStmt* pStmt, SMCreateStbReq* pReq) {
pReq->igExists = pStmt->ignoreExists;
pReq->delay1 = pStmt->pOptions->maxDelay1;
pReq->delay2 = pStmt->pOptions->maxDelay2;
pReq->watermark1 = pStmt->pOptions->watermark1;
pReq->watermark2 = pStmt->pOptions->watermark2;
// pReq->ttl = pStmt->pOptions->ttl;
columnDefNodeToField(pStmt->pCols, &pReq->pColumns);
columnDefNodeToField(pStmt->pTags, &pReq->pTags);
pReq->numOfColumns = LIST_LENGTH(pStmt->pCols);
......@@ -3523,6 +3558,7 @@ static int32_t buildCreateStbReq(STranslateContext* pCxt, SCreateTableStmt* pStm
} else {
pReq->commentLen = -1;
}
buildRollupFuncs(pStmt->pOptions->pRollupFuncs, &pReq->pFuncs);
SName tableName;
tNameExtractFullName(toName(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, &tableName), pReq->name);
......@@ -4296,7 +4332,10 @@ static int32_t translateShowCreateDatabase(STranslateContext* pCxt, SShowCreateD
}
static int32_t translateShowCreateTable(STranslateContext* pCxt, SShowCreateTableStmt* pStmt) {
return getTableMeta(pCxt, pStmt->dbName, pStmt->tableName, &pStmt->pMeta);
SName name;
toName(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, &name);
return getTableCfg(pCxt, &name, (STableCfg**)&pStmt->pCfg);
}
static int32_t translateQuery(STranslateContext* pCxt, SNode* pNode) {
......
source/libs/parser/src/parUtil.c
浏览文件 @ a23418d1
......@@ -561,6 +561,9 @@ int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalog
if (TSDB_CODE_SUCCESS == code) {
code = buildTableReq(pMetaCache->pTableIndex, &pCatalogReq->pTableIndex);
}
if (TSDB_CODE_SUCCESS == code) {
code = buildTableReq(pMetaCache->pTableCfg, &pCatalogReq->pTableCfg);
}
pCatalogReq->dNodeRequired = pMetaCache->dnodeRequired;
return code;
}
......@@ -657,6 +660,9 @@ int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMet
if (TSDB_CODE_SUCCESS == code) {
code = putTableDataToCache(pCatalogReq->pTableIndex, pMetaData->pTableIndex, &pMetaCache->pTableIndex);
}
if (TSDB_CODE_SUCCESS == code) {
code = putTableDataToCache(pCatalogReq->pTableCfg, pMetaData->pTableCfg, &pMetaCache->pTableCfg);
}
pMetaCache->pDnodes = pMetaData->pDnodeList;
return code;
}
......@@ -863,6 +869,10 @@ int32_t reserveTableIndexInCache(int32_t acctId, const char* pDb, const char* pT
return reserveTableReqInCache(acctId, pDb, pTable, &pMetaCache->pTableIndex);
}
int32_t reserveTableCfgInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
return reserveTableReqInCache(acctId, pDb, pTable, &pMetaCache->pTableCfg);
}
int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName, SArray** pIndexes) {
char fullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pName, fullName);
......@@ -877,6 +887,41 @@ int32_t getTableIndexFromCache(SParseMetaCache* pMetaCache, const SName* pName,
return code;
}
STableCfg* tableCfgDup(STableCfg* pCfg) {
STableCfg* pNew = taosMemoryMalloc(sizeof(*pNew));
memcpy(pNew, pCfg, sizeof(*pNew));
if (pNew->pComment) {
pNew->pComment = strdup(pNew->pComment);
}
if (pNew->pFuncs) {
pNew->pFuncs = taosArrayDup(pNew->pFuncs);
}
int32_t schemaSize = (pCfg->numOfColumns + pCfg->numOfTags) * sizeof(SSchema);
SSchema* pSchema = taosMemoryMalloc(schemaSize);
memcpy(pSchema, pCfg->pSchemas, schemaSize);
pNew->pSchemas = pSchema;
return pNew;
}
int32_t getTableCfgFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableCfg** pOutput) {
char fullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pName, fullName);
STableCfg* pCfg = NULL;
int32_t code = getMetaDataFromHash(fullName, strlen(fullName), pMetaCache->pTableCfg, (void**)&pCfg);
if (TSDB_CODE_SUCCESS == code) {
*pOutput = tableCfgDup(pCfg);
if (NULL == *pOutput) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
return code;
}
int32_t reserveDnodeRequiredInCache(SParseMetaCache* pMetaCache) {
pMetaCache->dnodeRequired = true;
return TSDB_CODE_SUCCESS;
......@@ -899,4 +944,5 @@ void destoryParseMetaCache(SParseMetaCache* pMetaCache) {
taosHashCleanup(pMetaCache->pUserAuth);
taosHashCleanup(pMetaCache->pUdf);
taosHashCleanup(pMetaCache->pTableIndex);
taosHashCleanup(pMetaCache->pTableCfg);
}
source/libs/parser/test/mockCatalog.cpp
浏览文件 @ a23418d1
......@@ -261,6 +261,11 @@ int32_t __catalogGetDnodeList(SCatalog* pCatalog, SRequestConnInfo* pConn, SArra
return g_mockCatalogService->catalogGetDnodeList(pDnodeList);
}
int32_t __catalogRefreshGetTableCfg(SCatalog* pCtg, SRequestConnInfo *pConn, const SName* pTableName, STableCfg** pCfg) {
*pCfg = (STableCfg*)taosMemoryCalloc(1, sizeof(STableCfg));
return 0;
}
void initMetaDataEnv() {
g_mockCatalogService.reset(new MockCatalogService());
......@@ -279,6 +284,7 @@ void initMetaDataEnv() {
stub.set(catalogRemoveTableMeta, __catalogRemoveTableMeta);
stub.set(catalogGetTableIndex, __catalogGetTableIndex);
stub.set(catalogGetDnodeList, __catalogGetDnodeList);
stub.set(catalogRefreshGetTableCfg, __catalogRefreshGetTableCfg);
// {
// AddrAny any("libcatalog.so");
// std::map<std::string,void*> result;
......
source/libs/parser/test/mockCatalogService.cpp
浏览文件 @ a23418d1
......@@ -199,6 +199,9 @@ class MockCatalogServiceImpl {
if (TSDB_CODE_SUCCESS == code && pCatalogReq->dNodeRequired) {
code = catalogGetDnodeList(&pMetaData->pDnodeList);
}
if (TSDB_CODE_SUCCESS == code) {
code = getAllTableCfg(pCatalogReq->pTableCfg, &pMetaData->pTableCfg);
}
return code;
}
......@@ -545,6 +548,20 @@ class MockCatalogServiceImpl {
return TSDB_CODE_SUCCESS;
}
int32_t getAllTableCfg(SArray* pTableCfgReq, SArray** pTableCfgData) const {
if (NULL != pTableCfgReq) {
int32_t ntables = taosArrayGetSize(pTableCfgReq);
*pTableCfgData = taosArrayInit(ntables, sizeof(SMetaRes));
for (int32_t i = 0; i < ntables; ++i) {
SMetaRes res = {0};
res.pRes = taosMemoryCalloc(1, sizeof(STableCfg));
res.code = TSDB_CODE_SUCCESS;
taosArrayPush(*pTableCfgData, &res);
}
}
return TSDB_CODE_SUCCESS;
}
uint64_t id_;
std::unique_ptr<TableBuilder> builder_;
DbMetaCache meta_;
......
source/libs/parser/test/parShowToUse.cpp
浏览文件 @ a23418d1
......@@ -45,7 +45,7 @@ TEST_F(ParserShowToUseTest, showCreateSTable) {
ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_SHOW_CREATE_STABLE_STMT);
ASSERT_EQ(pQuery->execMode, QUERY_EXEC_MODE_LOCAL);
ASSERT_TRUE(pQuery->haveResultSet);
ASSERT_NE(((SShowCreateTableStmt*)pQuery->pRoot)->pMeta, nullptr);
ASSERT_NE(((SShowCreateTableStmt*)pQuery->pRoot)->pCfg, nullptr);
});
run("SHOW CREATE STABLE st1");
......@@ -58,7 +58,7 @@ TEST_F(ParserShowToUseTest, showCreateTable) {
ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_SHOW_CREATE_TABLE_STMT);
ASSERT_EQ(pQuery->execMode, QUERY_EXEC_MODE_LOCAL);
ASSERT_TRUE(pQuery->haveResultSet);
ASSERT_NE(((SShowCreateTableStmt*)pQuery->pRoot)->pMeta, nullptr);
ASSERT_NE(((SShowCreateTableStmt*)pQuery->pRoot)->pCfg, nullptr);
});
run("SHOW CREATE TABLE t1");
......
source/libs/qcom/src/queryUtil.c
浏览文件 @ a23418d1
......@@ -19,6 +19,7 @@
#include "tmsg.h"
#include "trpc.h"
#include "tsched.h"
#include "cJSON.h"
#define VALIDNUMOFCOLS(x) ((x) >= TSDB_MIN_COLUMNS && (x) <= TSDB_MAX_COLUMNS)
#define VALIDNUMOFTAGS(x) ((x) >= 0 && (x) <= TSDB_MAX_TAGS)
......@@ -220,3 +221,209 @@ void destroyQueryExecRes(SQueryExecRes* pRes) {
qError("invalid exec result for request type %d", pRes->msgType);
}
}
int32_t dataConverToStr(char *str, int type, void *buf, int32_t bufSize, int32_t *len) {
int32_t n = 0;
switch (type) {
case TSDB_DATA_TYPE_NULL:
n = sprintf(str, "null");
break;
case TSDB_DATA_TYPE_BOOL:
n = sprintf(str, (*(int8_t*)buf) ? "true" : "false");
break;
case TSDB_DATA_TYPE_TINYINT:
n = sprintf(str, "%d", *(int8_t*)buf);
break;
case TSDB_DATA_TYPE_SMALLINT:
n = sprintf(str, "%d", *(int16_t*)buf);
break;
case TSDB_DATA_TYPE_INT:
n = sprintf(str, "%d", *(int32_t*)buf);
break;
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_TIMESTAMP:
n = sprintf(str, "%" PRId64, *(int64_t*)buf);
break;
case TSDB_DATA_TYPE_FLOAT:
n = sprintf(str, "%e", GET_FLOAT_VAL(buf));
break;
case TSDB_DATA_TYPE_DOUBLE:
n = sprintf(str, "%e", GET_DOUBLE_VAL(buf));
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
if (bufSize < 0) {
// tscError("invalid buf size");
return TSDB_CODE_TSC_INVALID_VALUE;
}
*str = '"';
memcpy(str + 1, buf, bufSize);
*(str + bufSize + 1) = '"';
n = bufSize + 2;
break;
case TSDB_DATA_TYPE_UTINYINT:
n = sprintf(str, "%d", *(uint8_t*)buf);
break;
case TSDB_DATA_TYPE_USMALLINT:
n = sprintf(str, "%d", *(uint16_t*)buf);
break;
case TSDB_DATA_TYPE_UINT:
n = sprintf(str, "%u", *(uint32_t*)buf);
break;
case TSDB_DATA_TYPE_UBIGINT:
n = sprintf(str, "%" PRIu64, *(uint64_t*)buf);
break;
default:
// tscError("unsupported type:%d", type);
return TSDB_CODE_TSC_INVALID_VALUE;
}
*len = n;
return TSDB_CODE_SUCCESS;
}
char* parseTagDatatoJson(void* p) {
char* string = NULL;
cJSON* json = cJSON_CreateObject();
if (json == NULL) {
goto end;
}
SArray* pTagVals = NULL;
if (tTagToValArray((const STag*)p, &pTagVals) != 0) {
goto end;
}
int16_t nCols = taosArrayGetSize(pTagVals);
char tagJsonKey[256] = {0};
for (int j = 0; j < nCols; ++j) {
STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, j);
// json key encode by binary
memset(tagJsonKey, 0, sizeof(tagJsonKey));
memcpy(tagJsonKey, pTagVal->pKey, strlen(pTagVal->pKey));
// json value
char type = pTagVal->type;
if (type == TSDB_DATA_TYPE_NULL) {
cJSON* value = cJSON_CreateNull();
if (value == NULL) {
goto end;
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else if (type == TSDB_DATA_TYPE_NCHAR) {
cJSON* value = NULL;
if (pTagVal->nData > 0) {
char* tagJsonValue = taosMemoryCalloc(pTagVal->nData, 1);
int32_t length = taosUcs4ToMbs((TdUcs4*)pTagVal->pData, pTagVal->nData, tagJsonValue);
if (length < 0) {
qError("charset:%s to %s. val:%s convert json value failed.", DEFAULT_UNICODE_ENCODEC, tsCharset,
pTagVal->pData);
taosMemoryFree(tagJsonValue);
goto end;
}
value = cJSON_CreateString(tagJsonValue);
taosMemoryFree(tagJsonValue);
if (value == NULL) {
goto end;
}
} else if (pTagVal->nData == 0) {
value = cJSON_CreateString("");
} else {
ASSERT(0);
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else if (type == TSDB_DATA_TYPE_DOUBLE) {
double jsonVd = *(double*)(&pTagVal->i64);
cJSON* value = cJSON_CreateNumber(jsonVd);
if (value == NULL) {
goto end;
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else if (type == TSDB_DATA_TYPE_BOOL) {
char jsonVd = *(char*)(&pTagVal->i64);
cJSON* value = cJSON_CreateBool(jsonVd);
if (value == NULL) {
goto end;
}
cJSON_AddItemToObject(json, tagJsonKey, value);
} else {
ASSERT(0);
}
}
string = cJSON_PrintUnformatted(json);
end:
cJSON_Delete(json);
return string;
}
int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
if (NULL == pSrc) {
*pDst = NULL;
return TSDB_CODE_SUCCESS;
}
int32_t metaSize = (pSrc->tableInfo.numOfColumns + pSrc->tableInfo.numOfTags) * sizeof(SSchema);
*pDst = taosMemoryMalloc(metaSize);
if (NULL == *pDst) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memcpy(*pDst, pSrc, metaSize);
return TSDB_CODE_SUCCESS;
}
int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst) {
if (NULL == pSrc) {
*pDst = NULL;
return TSDB_CODE_SUCCESS;
}
*pDst = taosMemoryMalloc(sizeof(*pSrc));
if (NULL == *pDst) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memcpy(*pDst, pSrc, sizeof(*pSrc));
if (pSrc->vgHash) {
(*pDst)->vgHash = taosHashInit(taosHashGetSize(pSrc->vgHash), taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
if (NULL == (*pDst)->vgHash) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
SVgroupInfo* vgInfo = NULL;
void *pIter = taosHashIterate(pSrc->vgHash, NULL);
while (pIter) {
vgInfo = pIter;
int32_t* vgId = taosHashGetKey(pIter, NULL);
if (0 != taosHashPut((*pDst)->vgHash, vgId, sizeof(*vgId), vgInfo, sizeof(*vgInfo))) {
qError("taosHashPut failed, vgId:%d", vgInfo->vgId);
taosHashCancelIterate(pSrc->vgHash, pIter);
taosHashCleanup((*pDst)->vgHash);
taosMemoryFreeClear(*pDst);
return TSDB_CODE_CTG_MEM_ERROR;
}
pIter = taosHashIterate(pSrc->vgHash, pIter);
}
}
return TSDB_CODE_SUCCESS;
}
source/libs/qcom/src/querymsg.c
浏览文件 @ a23418d1
......@@ -63,7 +63,7 @@ int32_t queryBuildUseDbOutput(SUseDbOutput *pOut, SUseDbRsp *usedbRsp) {
}
int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
SBuildTableMetaInput *pInput = input;
SBuildTableInput *pInput = input;
if (NULL == input || NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT;
}
......@@ -221,6 +221,27 @@ int32_t queryBuildGetTbIndexMsg(void *input, char **msg, int32_t msgSize, int32_
return TSDB_CODE_SUCCESS;
}
int32_t queryBuildGetTbCfgMsg(void *input, char **msg, int32_t msgSize, int32_t *msgLen, void*(*mallcFp)(int32_t)) {
if (NULL == msg || NULL == msgLen) {
return TSDB_CODE_TSC_INVALID_INPUT;
}
SBuildTableInput *pInput = input;
STableCfgReq cfgReq = {0};
cfgReq.header.vgId = pInput->vgId;
strcpy(cfgReq.dbFName, pInput->dbFName);
strcpy(cfgReq.tbName, pInput->tbName);
int32_t bufLen = tSerializeSTableCfgReq(NULL, 0, &cfgReq);
void *pBuf = (*mallcFp)(bufLen);
tSerializeSTableCfgReq(pBuf, bufLen, &cfgReq);
*msg = pBuf;
*msgLen = bufLen;
return TSDB_CODE_SUCCESS;
}
int32_t queryProcessUseDBRsp(void *output, char *msg, int32_t msgSize) {
SUseDbOutput *pOut = output;
......@@ -493,6 +514,21 @@ int32_t queryProcessGetTbIndexRsp(void *output, char *msg, int32_t msgSize) {
return TSDB_CODE_SUCCESS;
}
int32_t queryProcessGetTbCfgRsp(void *output, char *msg, int32_t msgSize) {
if (NULL == output || NULL == msg || msgSize <= 0) {
return TSDB_CODE_TSC_INVALID_INPUT;
}
STableCfgRsp *out = taosMemoryCalloc(1, sizeof(STableCfgRsp));
if (tDeserializeSTableCfgRsp(msg, msgSize, out) != 0) {
qError("tDeserializeSTableCfgRsp failed, msgSize:%d", msgSize);
return TSDB_CODE_INVALID_MSG;
}
*(STableCfgRsp**)output = out;
return TSDB_CODE_SUCCESS;
}
void initQueryModuleMsgHandle() {
queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryBuildTableMetaReqMsg;
......@@ -504,6 +540,8 @@ void initQueryModuleMsgHandle() {
queryBuildMsg[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)] = queryBuildRetrieveFuncMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)] = queryBuildGetUserAuthMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_TABLE_INDEX)] = queryBuildGetTbIndexMsg;
queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_CFG)] = queryBuildGetTbCfgMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_CFG)] = queryBuildGetTbCfgMsg;
queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryProcessTableMetaRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryProcessTableMetaRsp;
......@@ -514,6 +552,8 @@ void initQueryModuleMsgHandle() {
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)] = queryProcessRetrieveFuncRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)] = queryProcessGetUserAuthRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_TABLE_INDEX)] = queryProcessGetTbIndexRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_CFG)] = queryProcessGetTbCfgRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_CFG)] = queryProcessGetTbCfgRsp;
}
#pragma GCC diagnostic pop
source/util/src/terror.c
浏览文件 @ a23418d1
......@@ -134,6 +134,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_TSC_STMT_TBNAME_ERROR, "Stmt table name not s
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_STMT_CLAUSE_ERROR, "not supported stmt clause")
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_QUERY_KILLED, "Query killed")
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_NO_EXEC_NODE, "No available execution node")
TAOS_DEFINE_ERROR(TSDB_CODE_TSC_NOT_STABLE_ERROR, "Table is not a super table")
// mnode-common
TAOS_DEFINE_ERROR(TSDB_CODE_MND_APP_ERROR, "Mnode internal error")
......
tests/script/tsim/show/basic.sim
浏览文件 @ a23418d1
......@@ -36,6 +36,7 @@ sql create database db vgroups 3
sql use db
sql create table stb (ts timestamp, c int) tags (t int)
sql create table t0 using stb tags (0)
sql create table tba (ts timestamp, c1 binary(10), c2 nchar(10));
print =============== run show xxxx
sql show dnodes
......@@ -62,7 +63,7 @@ if $rows != 1 then
endi
#sql show streams,
sql show tables
if $rows != 1 then
if $rows != 2 then
return -1
endi
#sql show user_table_distributed
......@@ -98,7 +99,7 @@ if $rows != 1 then
endi
#sql select * from information_schema.`streams`
sql select * from information_schema.user_tables
if $rows != 29 then
if $rows != 30 then
return -1
endi
#sql select * from information_schema.user_table_distributed
......@@ -160,7 +161,7 @@ if $rows != 1 then
endi
#sql show streams,
sql show tables
if $rows != 1 then
if $rows != 2 then
return -1
endi
#sql show user_table_distributed
......@@ -196,7 +197,7 @@ if $rows != 1 then
endi
#sql select * from performance_schema.`streams`
sql select * from information_schema.user_tables
if $rows != 29 then
if $rows != 30 then
return -1
endi
#sql select * from information_schema.user_table_distributed
......@@ -209,5 +210,22 @@ if $rows != 3 then
return -1
endi
sql show create stable stb;
if $rows != 1 then
return -1
endi
sql show create table t0;
if $rows != 1 then
return -1
endi
sql show create table tba;
if $rows != 1 then
return -1
endi
sql_error show create stable t0;
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或