diff --git a/examples/c/tmq.c b/examples/c/tmq.c index 5d7f1bbe70..94e0b86821 100644 --- a/examples/c/tmq.c +++ b/examples/c/tmq.c @@ -30,21 +30,35 @@ static void msg_process(TAOS_RES* msg) { if (tmq_get_res_type(msg) == TMQ_RES_TABLE_META) { tmq_raw_data *raw = tmq_get_raw_meta(msg); if(raw){ - TAOS* pConn = taos_connect("192.168.1.86", "root", "taosdata", "abc1", 0); + TAOS* pConn = taos_connect("192.168.1.86", "root", "taosdata", NULL, 0); if (pConn == NULL) { return; } + + TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 5"); + if (taos_errno(pRes) != 0) { + printf("error in create db, reason:%s\n", taos_errstr(pRes)); + return; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "use abc1"); + if (taos_errno(pRes) != 0) { + printf("error in use db, reason:%s\n", taos_errstr(pRes)); + return; + } + taos_free_result(pRes); + int32_t ret = taos_write_raw_meta(pConn, raw); printf("write raw data: %s\n", tmq_err2str(ret)); - free(raw); taos_close(pConn); } + tmq_free_raw_meta(raw); char* result = tmq_get_json_meta(msg); if(result){ printf("meta result: %s\n", result); - free(result); } - printf("meta:%p\n", raw); + tmq_free_json_meta(result); return; } while (1) { @@ -68,7 +82,7 @@ int32_t init_env() { return -1; } - TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1"); + TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 5"); if (taos_errno(pRes) != 0) { printf("error in create db, reason:%s\n", taos_errstr(pRes)); return -1; @@ -82,15 +96,14 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = - taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int)"); + pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 nchar(8), t4 bool)"); if (taos_errno(pRes) != 0) { printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000)"); + pRes = taos_query(pConn, "create table if not exists ct0 using st1 tags(1000, \"ttt\", true)"); if (taos_errno(pRes) != 0) { printf("failed to create child table tu1, reason:%s\n", taos_errstr(pRes)); return -1; @@ -104,13 +117,20 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists ct1 using st1 tags(2000)"); + pRes = taos_query(pConn, "create table if not exists ct1 using st1(t1) tags(2000)"); if (taos_errno(pRes) != 0) { printf("failed to create child table ct1, reason:%s\n", taos_errstr(pRes)); return -1; } taos_free_result(pRes); + pRes = taos_query(pConn, "create table if not exists ct2 using st1(t1) tags(NULL)"); + if (taos_errno(pRes) != 0) { + printf("failed to create child table ct2, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + pRes = taos_query(pConn, "insert into ct1 values(now, 3, 4, 'b')"); if (taos_errno(pRes) != 0) { printf("failed to insert into ct1, reason:%s\n", taos_errstr(pRes)); @@ -118,7 +138,7 @@ int32_t init_env() { } taos_free_result(pRes); - pRes = taos_query(pConn, "create table if not exists ct3 using st1 tags(3000)"); + pRes = taos_query(pConn, "create table if not exists ct3 using st1(t1) tags(3000)"); if (taos_errno(pRes) != 0) { printf("failed to create child table ct3, reason:%s\n", taos_errstr(pRes)); return -1; @@ -202,6 +222,13 @@ int32_t init_env() { } taos_free_result(pRes); + pRes = taos_query(pConn, "alter table n1 comment 'hello'"); + if (taos_errno(pRes) != 0) { + printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + pRes = taos_query(pConn, "alter table n1 drop column c1"); if (taos_errno(pRes) != 0) { printf("failed to alter normal table n1, reason:%s\n", taos_errstr(pRes)); @@ -230,6 +257,27 @@ int32_t init_env() { } taos_free_result(pRes); + pRes = taos_query(pConn, "create table jt2 using jt tags('')"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table jt2, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(16)) tags(t1 int, t3 nchar(8), t4 bool)"); + if (taos_errno(pRes) != 0) { + printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + + pRes = taos_query(pConn, "drop table st1"); + if (taos_errno(pRes) != 0) { + printf("failed to drop super table st1, reason:%s\n", taos_errstr(pRes)); + return -1; + } + taos_free_result(pRes); + return 0; } diff --git a/include/client/taos.h b/include/client/taos.h index 362782b420..690c473986 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -265,7 +265,9 @@ typedef struct tmq_raw_data tmq_raw_data; DLL_EXPORT tmq_res_t tmq_get_res_type(TAOS_RES *res); DLL_EXPORT tmq_raw_data *tmq_get_raw_meta(TAOS_RES *res); DLL_EXPORT int32_t taos_write_raw_meta(TAOS *taos, tmq_raw_data *raw_meta); -DLL_EXPORT char *tmq_get_json_meta(TAOS_RES *res); // Returning null means error. Returned result need to be freed. +DLL_EXPORT void tmq_free_raw_meta(tmq_raw_data *rawMeta); +DLL_EXPORT char *tmq_get_json_meta(TAOS_RES *res); // Returning null means error. Returned result need to be freed by tmq_free_json_meta +DLL_EXPORT void tmq_free_json_meta(char* jsonMeta); DLL_EXPORT const char *tmq_get_topic_name(TAOS_RES *res); DLL_EXPORT const char *tmq_get_db_name(TAOS_RES *res); DLL_EXPORT int32_t tmq_get_vgroup_id(TAOS_RES *res); diff --git a/include/common/tcommon.h b/include/common/tcommon.h index d8264ac5b5..f672d47c92 100644 --- a/include/common/tcommon.h +++ b/include/common/tcommon.h @@ -40,7 +40,6 @@ enum { || x == TDMT_VND_CREATE_TABLE \ || x == TDMT_VND_ALTER_TABLE \ || x == TDMT_VND_DROP_TABLE \ - || x == TDMT_VND_DROP_TTL_TABLE \ ) // clang-format on diff --git a/include/common/tdataformat.h b/include/common/tdataformat.h index 1011b90ce8..311eb72b4d 100644 --- a/include/common/tdataformat.h +++ b/include/common/tdataformat.h @@ -171,6 +171,7 @@ struct SColVal { #pragma pack(push, 1) struct STagVal { +// char colName[TSDB_COL_NAME_LEN]; // only used for tmq_get_meta union { int16_t cid; char *pKey; diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 0ab65e7978..c8e13fce3d 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1939,6 +1939,8 @@ typedef struct SVCreateStbReq { SSchemaWrapper schemaRow; SSchemaWrapper schemaTag; SRSmaParam rsmaParam; + int32_t alterOriDataLen; + void* alterOriData; } SVCreateStbReq; int tEncodeSVCreateStbReq(SEncoder* pCoder, const SVCreateStbReq* pReq); @@ -1966,7 +1968,9 @@ typedef struct SVCreateTbReq { int8_t type; union { struct { + char* name; // super table name tb_uid_t suid; + SArray* tagName; uint8_t* pTag; } ctb; struct { @@ -1983,6 +1987,9 @@ static FORCE_INLINE void tdDestroySVCreateTbReq(SVCreateTbReq* req) { taosMemoryFreeClear(req->comment); if (req->type == TSDB_CHILD_TABLE) { taosMemoryFreeClear(req->ctb.pTag); + taosMemoryFreeClear(req->ctb.name); + taosArrayDestroy(req->ctb.tagName); + req->ctb.tagName = NULL; } else if (req->type == TSDB_NORMAL_TABLE) { taosMemoryFreeClear(req->ntb.schemaRow.pSchema); } @@ -2066,12 +2073,14 @@ typedef struct { int32_t bytes; // TSDB_ALTER_TABLE_DROP_COLUMN // TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES + int8_t colModType; int32_t colModBytes; // TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME char* colNewName; // TSDB_ALTER_TABLE_UPDATE_TAG_VAL char* tagName; int8_t isNull; + int8_t tagType; uint32_t nTagVal; uint8_t* pTagVal; // TSDB_ALTER_TABLE_UPDATE_OPTIONS @@ -2858,8 +2867,8 @@ typedef struct { static FORCE_INLINE int32_t tEncodeSMqMetaRsp(void** buf, const SMqMetaRsp* pRsp) { int32_t tlen = 0; - // tlen += taosEncodeFixedI64(buf, pRsp->reqOffset); - // tlen += taosEncodeFixedI64(buf, pRsp->rspOffset); + tlen += taosEncodeFixedI64(buf, pRsp->reqOffset); + tlen += taosEncodeFixedI64(buf, pRsp->rspOffset); tlen += taosEncodeFixedI16(buf, pRsp->resMsgType); tlen += taosEncodeFixedI32(buf, pRsp->metaRspLen); tlen += taosEncodeBinary(buf, pRsp->metaRsp, pRsp->metaRspLen); @@ -2867,8 +2876,7 @@ static FORCE_INLINE int32_t tEncodeSMqMetaRsp(void** buf, const SMqMetaRsp* pRsp } static FORCE_INLINE void* tDecodeSMqMetaRsp(const void* buf, SMqMetaRsp* pRsp) { - // buf = taosDecodeFixedI64(buf, &pRsp->reqOffset); - // buf = taosDecodeFixedI64(buf, &pRsp->rspOffset); + buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);buf = taosDecodeFixedI64(buf, &pRsp->rspOffset); buf = taosDecodeFixedI16(buf, &pRsp->resMsgType); buf = taosDecodeFixedI32(buf, &pRsp->metaRspLen); buf = taosDecodeBinary(buf, &pRsp->metaRsp, pRsp->metaRspLen); diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index a4aec72d4f..a3de9164a2 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -29,7 +29,7 @@ struct SMetaData; typedef struct SStmtCallback { TAOS_STMT* pStmt; int32_t (*getTbNameFn)(TAOS_STMT*, char**); - int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, char*, bool, SHashObj*, SHashObj*); + int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, char*, bool, SHashObj*, SHashObj*, const char*); int32_t (*getExecInfoFn)(TAOS_STMT*, SHashObj**, SHashObj**); } SStmtCallback; @@ -84,7 +84,7 @@ int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBu int32_t rowNum); int32_t qBuildStmtColFields(void* pDataBlock, int32_t* fieldNum, TAOS_FIELD_E** fields); int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD_E** fields); -int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tName, TAOS_MULTI_BIND* bind, +int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const char* sTableName, char* tName, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen); void destroyBoundColumnInfo(void* pBoundInfo); int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf, @@ -93,7 +93,7 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* void* smlInitHandle(SQuery* pQuery); void smlDestroyHandle(void* pHandle); int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta, - char* tableName, char* msgBuf, int16_t msgBufLen); + char* tableName, const char* sTableName, int32_t sTableNameLen, char* msgBuf, int16_t msgBufLen); int32_t smlBuildOutput(void* handle, SHashObj* pVgHash); int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray); diff --git a/include/util/tencode.h b/include/util/tencode.h index e318d4f240..ad642cd612 100644 --- a/include/util/tencode.h +++ b/include/util/tencode.h @@ -440,7 +440,7 @@ static FORCE_INLINE bool tDecodeIsEnd(SDecoder* pCoder) { return (pCoder->size = static FORCE_INLINE void* tEncoderMalloc(SEncoder* pCoder, int32_t size) { void* p = NULL; - SCoderMem* pMem = (SCoderMem*)taosMemoryMalloc(sizeof(*pMem) + size); + SCoderMem* pMem = (SCoderMem*)taosMemoryCalloc(1, sizeof(*pMem) + size); if (pMem) { pMem->next = pCoder->mList; pCoder->mList = pMem; @@ -451,7 +451,7 @@ static FORCE_INLINE void* tEncoderMalloc(SEncoder* pCoder, int32_t size) { static FORCE_INLINE void* tDecoderMalloc(SDecoder* pCoder, int32_t size) { void* p = NULL; - SCoderMem* pMem = (SCoderMem*)taosMemoryMalloc(sizeof(*pMem) + size); + SCoderMem* pMem = (SCoderMem*)taosMemoryCalloc(1, sizeof(*pMem) + size); if (pMem) { pMem->next = pCoder->mList; pCoder->mList = pMem; diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index 6e689adf95..9ac34cf2df 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -2256,7 +2256,7 @@ static int32_t smlInsertData(SSmlHandle *info) { (*pMeta)->tableMeta->uid = tableData->uid; // one table merge data block together according uid code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat, - (*pMeta)->tableMeta, tableData->childTableName, info->msgBuf.buf, info->msgBuf.len); + (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen, info->msgBuf.buf, info->msgBuf.len); if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%" PRIx64 " smlBindData failed", info->id); return code; diff --git a/source/client/src/clientStmt.c b/source/client/src/clientStmt.c index 92ed830dee..bf00965c7a 100644 --- a/source/client/src/clientStmt.c +++ b/source/client/src/clientStmt.c @@ -136,7 +136,7 @@ int32_t stmtRestoreQueryFields(STscStmt* pStmt) { return TSDB_CODE_SUCCESS; } -int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName) { +int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, const char* sTableName) { STscStmt* pStmt = (STscStmt*)stmt; strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1); @@ -147,6 +147,7 @@ int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, pStmt->bInfo.tbType = pTableMeta->tableType; pStmt->bInfo.boundTags = tags; pStmt->bInfo.tagsCached = false; + strcpy(pStmt->bInfo.stbFName, sTableName); return TSDB_CODE_SUCCESS; } @@ -162,10 +163,10 @@ int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockH } int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, bool autoCreateTbl, - SHashObj* pVgHash, SHashObj* pBlockHash) { + SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) { STscStmt* pStmt = (STscStmt*)stmt; - STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName)); + STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName, sTableName)); STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl)); pStmt->sql.autoCreateTbl = autoCreateTbl; @@ -253,7 +254,7 @@ int32_t stmtCleanBindInfo(STscStmt* pStmt) { destroyBoundColumnInfo(pStmt->bInfo.boundTags); taosMemoryFreeClear(pStmt->bInfo.boundTags); } - + memset(pStmt->bInfo.stbFName, 0, TSDB_TABLE_FNAME_LEN); return TSDB_CODE_SUCCESS; } @@ -592,7 +593,7 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) { } tscDebug("start to bind stmt tag values"); - STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.sname.tname, + STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.stbFName, pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen)); return TSDB_CODE_SUCCESS; diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c index b0542e350f..ad47348c22 100644 --- a/source/client/src/tmq.c +++ b/source/client/src/tmq.c @@ -1858,9 +1858,9 @@ const char* tmq_get_table_name(TAOS_RES* res) { return NULL; } -tmq_raw_data* tmq_get_raw_meta(TAOS_RES* res) { +tmq_raw_data *tmq_get_raw_meta(TAOS_RES* res) { if (TD_RES_TMQ_META(res)) { - tmq_raw_data* raw = taosMemoryCalloc(1, sizeof(tmq_raw_data)); + tmq_raw_data *raw = taosMemoryCalloc(1, sizeof(tmq_raw_data)); SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res; raw->raw_meta = pMetaRspObj->metaRsp.metaRsp; raw->raw_meta_len = pMetaRspObj->metaRsp.metaRspLen; @@ -1870,8 +1870,7 @@ tmq_raw_data* tmq_get_raw_meta(TAOS_RES* res) { return NULL; } -static char* buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* schemaTag, char* name, int64_t id, - int8_t t) { +static char *buildCreateTableJson(SSchemaWrapper *schemaRow, SSchemaWrapper* schemaTag, char* name, int64_t id, int8_t t){ char* string = NULL; cJSON* json = cJSON_CreateObject(); if (json == NULL) { @@ -1880,32 +1879,32 @@ static char* buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* sch cJSON* type = cJSON_CreateString("create"); cJSON_AddItemToObject(json, "type", type); - char uid[32] = {0}; - sprintf(uid, "%" PRIi64, id); - cJSON* id_ = cJSON_CreateString(uid); - cJSON_AddItemToObject(json, "id", id_); +// char uid[32] = {0}; +// sprintf(uid, "%"PRIi64, id); +// cJSON* id_ = cJSON_CreateString(uid); +// cJSON_AddItemToObject(json, "id", id_); cJSON* tableName = cJSON_CreateString(name); cJSON_AddItemToObject(json, "tableName", tableName); cJSON* tableType = cJSON_CreateString(t == TSDB_NORMAL_TABLE ? "normal" : "super"); cJSON_AddItemToObject(json, "tableType", tableType); - // cJSON* version = cJSON_CreateNumber(1); - // cJSON_AddItemToObject(json, "version", version); +// cJSON* version = cJSON_CreateNumber(1); +// cJSON_AddItemToObject(json, "version", version); cJSON* columns = cJSON_CreateArray(); - for (int i = 0; i < schemaRow->nCols; i++) { - cJSON* column = cJSON_CreateObject(); - SSchema* s = schemaRow->pSchema + i; - cJSON* cname = cJSON_CreateString(s->name); + for(int i = 0; i < schemaRow->nCols; i++){ + cJSON* column = cJSON_CreateObject(); + SSchema *s = schemaRow->pSchema + i; + cJSON* cname = cJSON_CreateString(s->name); cJSON_AddItemToObject(column, "name", cname); cJSON* ctype = cJSON_CreateNumber(s->type); cJSON_AddItemToObject(column, "type", ctype); - if (s->type == TSDB_DATA_TYPE_BINARY) { + if(s->type == TSDB_DATA_TYPE_BINARY){ int32_t length = s->bytes - VARSTR_HEADER_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(column, "length", cbytes); - } else if (s->type == TSDB_DATA_TYPE_NCHAR) { - int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + }else if (s->type == TSDB_DATA_TYPE_NCHAR){ + int32_t length = (s->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(column, "length", cbytes); } cJSON_AddItemToArray(columns, column); @@ -1913,20 +1912,20 @@ static char* buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* sch cJSON_AddItemToObject(json, "columns", columns); cJSON* tags = cJSON_CreateArray(); - for (int i = 0; schemaTag && i < schemaTag->nCols; i++) { - cJSON* tag = cJSON_CreateObject(); - SSchema* s = schemaTag->pSchema + i; - cJSON* tname = cJSON_CreateString(s->name); + for(int i = 0; schemaTag && i < schemaTag->nCols; i++){ + cJSON* tag = cJSON_CreateObject(); + SSchema *s = schemaTag->pSchema + i; + cJSON* tname = cJSON_CreateString(s->name); cJSON_AddItemToObject(tag, "name", tname); cJSON* ttype = cJSON_CreateNumber(s->type); cJSON_AddItemToObject(tag, "type", ttype); - if (s->type == TSDB_DATA_TYPE_BINARY) { + if(s->type == TSDB_DATA_TYPE_BINARY){ int32_t length = s->bytes - VARSTR_HEADER_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(tag, "length", cbytes); - } else if (s->type == TSDB_DATA_TYPE_NCHAR) { - int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + }else if (s->type == TSDB_DATA_TYPE_NCHAR){ + int32_t length = (s->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(tag, "length", cbytes); } cJSON_AddItemToArray(tags, tag); @@ -1938,13 +1937,105 @@ static char* buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* sch return string; } -static char* processCreateStb(SMqMetaRsp* metaRsp) { +static char *buildAlterSTableJson(void* alterData, int32_t alterDataLen){ + SMAlterStbReq req = {0}; + cJSON* json = NULL; + char* string = NULL; + + if (tDeserializeSMAlterStbReq(alterData, alterDataLen, &req) != 0) { + goto end; + } + + json = cJSON_CreateObject(); + if (json == NULL) { + goto end; + } + cJSON* type = cJSON_CreateString("alter"); + cJSON_AddItemToObject(json, "type", type); +// cJSON* uid = cJSON_CreateNumber(id); +// cJSON_AddItemToObject(json, "uid", uid); + SName name = {0}; + tNameFromString(&name, req.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE); + cJSON* tableName = cJSON_CreateString(name.tname); + cJSON_AddItemToObject(json, "tableName", tableName); + cJSON* tableType = cJSON_CreateString("super"); + cJSON_AddItemToObject(json, "tableType", tableType); + + cJSON* alterType = cJSON_CreateNumber(req.alterType); + cJSON_AddItemToObject(json, "alterType", alterType); + switch (req.alterType) { + case TSDB_ALTER_TABLE_ADD_TAG: + case TSDB_ALTER_TABLE_ADD_COLUMN: { + TAOS_FIELD *field = taosArrayGet(req.pFields, 0); + cJSON* colName = cJSON_CreateString(field->name); + cJSON_AddItemToObject(json, "colName", colName); + cJSON* colType = cJSON_CreateNumber(field->type); + cJSON_AddItemToObject(json, "colType", colType); + + if(field->type == TSDB_DATA_TYPE_BINARY){ + int32_t length = field->bytes - VARSTR_HEADER_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); + cJSON_AddItemToObject(json, "colLength", cbytes); + }else if (field->type == TSDB_DATA_TYPE_NCHAR){ + int32_t length = (field->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); + cJSON_AddItemToObject(json, "colLength", cbytes); + } + break; + } + case TSDB_ALTER_TABLE_DROP_TAG: + case TSDB_ALTER_TABLE_DROP_COLUMN:{ + TAOS_FIELD *field = taosArrayGet(req.pFields, 0); + cJSON* colName = cJSON_CreateString(field->name); + cJSON_AddItemToObject(json, "colName", colName); + break; + } + case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES: + case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:{ + TAOS_FIELD *field = taosArrayGet(req.pFields, 0); + cJSON* colName = cJSON_CreateString(field->name); + cJSON_AddItemToObject(json, "colName", colName); + cJSON* colType = cJSON_CreateNumber(field->type); + cJSON_AddItemToObject(json, "colType", colType); + if(field->type == TSDB_DATA_TYPE_BINARY){ + int32_t length = field->bytes - VARSTR_HEADER_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); + cJSON_AddItemToObject(json, "colLength", cbytes); + }else if (field->type == TSDB_DATA_TYPE_NCHAR){ + int32_t length = (field->bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); + cJSON_AddItemToObject(json, "colLength", cbytes); + } + break; + } + case TSDB_ALTER_TABLE_UPDATE_TAG_NAME: + case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:{ + TAOS_FIELD *oldField = taosArrayGet(req.pFields, 0); + TAOS_FIELD *newField = taosArrayGet(req.pFields, 1); + cJSON* colName = cJSON_CreateString(oldField->name); + cJSON_AddItemToObject(json, "colName", colName); + cJSON* colNewName = cJSON_CreateString(newField->name); + cJSON_AddItemToObject(json, "colNewName", colNewName); + break; + } + default: + break; + } + string = cJSON_PrintUnformatted(json); + + end: + cJSON_Delete(json); + tFreeSMAltertbReq(&req); + return string; +} + +static char *processCreateStb(SMqMetaRsp *metaRsp){ SVCreateStbReq req = {0}; SDecoder coder; - char* string = NULL; + char* string = NULL; // decode and process req - void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead); tDecoderInit(&coder, data, len); @@ -1955,68 +2046,95 @@ static char* processCreateStb(SMqMetaRsp* metaRsp) { tDecoderClear(&coder); return string; -_err: + _err: tDecoderClear(&coder); return string; } -static char* buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t id) { +static char *processAlterStb(SMqMetaRsp *metaRsp){ + SVCreateStbReq req = {0}; + SDecoder coder; char* string = NULL; + + // decode and process req + void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); + int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead); + tDecoderInit(&coder, data, len); + + if (tDecodeSVCreateStbReq(&coder, &req) < 0) { + goto _err; + } + string = buildAlterSTableJson(req.alterOriData, req.alterOriDataLen); + tDecoderClear(&coder); + return string; + + _err: + tDecoderClear(&coder); + return string; +} + +static char *buildCreateCTableJson(STag* pTag, char* sname, char* name, SArray* tagName, int64_t id){ + char* string = NULL; + SArray* pTagVals = NULL; cJSON* json = cJSON_CreateObject(); if (json == NULL) { return string; } cJSON* type = cJSON_CreateString("create"); cJSON_AddItemToObject(json, "type", type); - char cid[32] = {0}; - sprintf(cid, "%" PRIi64, id); - cJSON* cid_ = cJSON_CreateString(cid); - cJSON_AddItemToObject(json, "id", cid_); +// char cid[32] = {0}; +// sprintf(cid, "%"PRIi64, id); +// cJSON* cid_ = cJSON_CreateString(cid); +// cJSON_AddItemToObject(json, "id", cid_); cJSON* tableName = cJSON_CreateString(name); cJSON_AddItemToObject(json, "tableName", tableName); cJSON* tableType = cJSON_CreateString("child"); cJSON_AddItemToObject(json, "tableType", tableType); - - char sid_[32] = {0}; - sprintf(sid_, "%" PRIi64, sid); - cJSON* using = cJSON_CreateString(sid_); + cJSON* using = cJSON_CreateString(sname); cJSON_AddItemToObject(json, "using", using); - // cJSON* version = cJSON_CreateNumber(1); - // cJSON_AddItemToObject(json, "version", version); +// cJSON* version = cJSON_CreateNumber(1); +// cJSON_AddItemToObject(json, "version", version); cJSON* tags = cJSON_CreateArray(); - - if (tTagIsJson(pTag)) { // todo - char* pJson = parseTagDatatoJson(pTag); - - cJSON* tag = cJSON_CreateObject(); - cJSON* tname = cJSON_CreateString("unknown"); // todo - cJSON_AddItemToObject(tag, "name", tname); - cJSON* ttype = cJSON_CreateNumber(TSDB_DATA_TYPE_JSON); - cJSON_AddItemToObject(tag, "type", ttype); - cJSON* tvalue = cJSON_CreateString(pJson); - cJSON_AddItemToObject(tag, "value", tvalue); - cJSON_AddItemToArray(tags, tag); - cJSON_AddItemToObject(json, "tags", tags); - - string = cJSON_PrintUnformatted(json); - goto end; - } - - SArray* pTagVals = NULL; int32_t code = tTagToValArray(pTag, &pTagVals); if (code) { goto end; } - for (int i = 0; i < taosArrayGetSize(pTagVals); i++) { + if (tTagIsJson(pTag)) { + STag* p = (STag*)pTag; + if(p->nTag == 0){ + goto end; + } + char* pJson = parseTagDatatoJson(pTag); + cJSON* tag = cJSON_CreateObject(); + STagVal* pTagVal = taosArrayGet(pTagVals, 0); + + char* ptname = taosArrayGet(tagName, 0); + cJSON* tname = cJSON_CreateString(ptname); + cJSON_AddItemToObject(tag, "name", tname); +// cJSON* cid_ = cJSON_CreateString(""); +// cJSON_AddItemToObject(tag, "cid", cid_); + cJSON* ttype = cJSON_CreateNumber(TSDB_DATA_TYPE_JSON); + cJSON_AddItemToObject(tag, "type", ttype); + cJSON* tvalue = cJSON_CreateString(pJson); + cJSON_AddItemToObject(tag, "value", tvalue); + cJSON_AddItemToArray(tags, tag); + taosMemoryFree(pJson); + goto end; + } + + for(int i = 0; i < taosArrayGetSize(pTagVals); i++){ STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, i); cJSON* tag = cJSON_CreateObject(); - // cJSON* tname = cJSON_CreateNumber(pTagVal->cid); - cJSON* tname = cJSON_CreateString("unkonwn"); // todo + + char* ptname = taosArrayGet(tagName, i); + cJSON* tname = cJSON_CreateString(ptname); cJSON_AddItemToObject(tag, "name", tname); +// cJSON* cid = cJSON_CreateNumber(pTagVal->cid); +// cJSON_AddItemToObject(tag, "cid", cid); cJSON* ttype = cJSON_CreateNumber(pTagVal->type); cJSON_AddItemToObject(tag, "type", ttype); @@ -2034,22 +2152,22 @@ static char* buildCreateCTableJson(STag* pTag, int64_t sid, char* name, int64_t cJSON_AddItemToObject(tag, "value", tvalue); cJSON_AddItemToArray(tags, tag); } + + end: cJSON_AddItemToObject(json, "tags", tags); string = cJSON_PrintUnformatted(json); - -end: - cJSON_Delete(json); + taosArrayDestroy(pTagVals); return string; } -static char* processCreateTable(SMqMetaRsp* metaRsp) { +static char *processCreateTable(SMqMetaRsp *metaRsp){ SDecoder decoder = {0}; SVCreateTbBatchReq req = {0}; - SVCreateTbReq* pCreateReq; - char* string = NULL; + SVCreateTbReq *pCreateReq; + char *string = NULL; // decode - void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead); tDecoderInit(&decoder, data, len); if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) { @@ -2059,29 +2177,27 @@ static char* processCreateTable(SMqMetaRsp* metaRsp) { // loop to create table for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { pCreateReq = req.pReqs + iReq; - if (pCreateReq->type == TSDB_CHILD_TABLE) { - string = - buildCreateCTableJson((STag*)pCreateReq->ctb.pTag, pCreateReq->ctb.suid, pCreateReq->name, pCreateReq->uid); - } else if (pCreateReq->type == TSDB_NORMAL_TABLE) { - string = - buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE); + if(pCreateReq->type == TSDB_CHILD_TABLE){ + string = buildCreateCTableJson((STag*)pCreateReq->ctb.pTag, pCreateReq->ctb.name, pCreateReq->name, pCreateReq->ctb.tagName, pCreateReq->uid); + }else if(pCreateReq->type == TSDB_NORMAL_TABLE){ + string = buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE); } } tDecoderClear(&decoder); -_exit: + _exit: tDecoderClear(&decoder); return string; } -static char* processAlterTable(SMqMetaRsp* metaRsp) { - SDecoder decoder = {0}; - SVAlterTbReq vAlterTbReq = {0}; - char* string = NULL; +static char *processAlterTable(SMqMetaRsp *metaRsp){ + SDecoder decoder = {0}; + SVAlterTbReq vAlterTbReq = {0}; + char *string = NULL; // decode - void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead); tDecoderInit(&decoder, data, len); if (tDecodeSVAlterTbReq(&decoder, &vAlterTbReq) < 0) { @@ -2094,76 +2210,88 @@ static char* processAlterTable(SMqMetaRsp* metaRsp) { } cJSON* type = cJSON_CreateString("alter"); cJSON_AddItemToObject(json, "type", type); - // cJSON* uid = cJSON_CreateNumber(id); - // cJSON_AddItemToObject(json, "uid", uid); +// cJSON* uid = cJSON_CreateNumber(id); +// cJSON_AddItemToObject(json, "uid", uid); cJSON* tableName = cJSON_CreateString(vAlterTbReq.tbName); cJSON_AddItemToObject(json, "tableName", tableName); cJSON* tableType = cJSON_CreateString(vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_TAG_VAL ? "child" : "normal"); cJSON_AddItemToObject(json, "tableType", tableType); + cJSON* alterType = cJSON_CreateNumber(vAlterTbReq.action); + cJSON_AddItemToObject(json, "alterType", alterType); switch (vAlterTbReq.action) { case TSDB_ALTER_TABLE_ADD_COLUMN: { - cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_ADD_COLUMN); - cJSON_AddItemToObject(json, "alterType", alterType); cJSON* colName = cJSON_CreateString(vAlterTbReq.colName); cJSON_AddItemToObject(json, "colName", colName); cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type); cJSON_AddItemToObject(json, "colType", colType); - if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY) { + if(vAlterTbReq.type == TSDB_DATA_TYPE_BINARY){ int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(json, "colLength", cbytes); - } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) { - int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + }else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR){ + int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(json, "colLength", cbytes); } break; } - case TSDB_ALTER_TABLE_DROP_COLUMN: { - cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_DROP_COLUMN); - cJSON_AddItemToObject(json, "alterType", alterType); + case TSDB_ALTER_TABLE_DROP_COLUMN:{ cJSON* colName = cJSON_CreateString(vAlterTbReq.colName); cJSON_AddItemToObject(json, "colName", colName); break; } - case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: { - cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES); - cJSON_AddItemToObject(json, "alterType", alterType); + case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:{ cJSON* colName = cJSON_CreateString(vAlterTbReq.colName); cJSON_AddItemToObject(json, "colName", colName); - cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type); + cJSON* colType = cJSON_CreateNumber(vAlterTbReq.colModType); cJSON_AddItemToObject(json, "colType", colType); - if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY) { - int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + if(vAlterTbReq.colModType == TSDB_DATA_TYPE_BINARY){ + int32_t length = vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(json, "colLength", cbytes); - } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) { - int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE; - cJSON* cbytes = cJSON_CreateNumber(length); + }else if (vAlterTbReq.colModType == TSDB_DATA_TYPE_NCHAR){ + int32_t length = (vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE; + cJSON* cbytes = cJSON_CreateNumber(length); cJSON_AddItemToObject(json, "colLength", cbytes); } break; } - case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: { - cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME); - cJSON_AddItemToObject(json, "alterType", alterType); + case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:{ cJSON* colName = cJSON_CreateString(vAlterTbReq.colName); cJSON_AddItemToObject(json, "colName", colName); cJSON* colNewName = cJSON_CreateString(vAlterTbReq.colNewName); cJSON_AddItemToObject(json, "colNewName", colNewName); break; } - case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: { - cJSON* alterType = cJSON_CreateNumber(TSDB_ALTER_TABLE_UPDATE_TAG_VAL); - cJSON_AddItemToObject(json, "alterType", alterType); + case TSDB_ALTER_TABLE_UPDATE_TAG_VAL:{ cJSON* tagName = cJSON_CreateString(vAlterTbReq.tagName); cJSON_AddItemToObject(json, "colName", tagName); - cJSON* colValue = cJSON_CreateString("invalid, todo"); // todo - cJSON_AddItemToObject(json, "colValue", colValue); - cJSON* isNull = cJSON_CreateBool(vAlterTbReq.isNull); - cJSON_AddItemToObject(json, "colValueNull", isNull); + + bool isNull = vAlterTbReq.isNull; + if(vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON){ + STag *jsonTag = (STag *)vAlterTbReq.pTagVal; + if(jsonTag->nTag == 0) isNull = true; + } + if (!isNull){ + char* buf = NULL; + + if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) { + ASSERT(tTagIsJson(vAlterTbReq.pTagVal) == true); + buf = parseTagDatatoJson(vAlterTbReq.pTagVal); + } else { + buf = taosMemoryCalloc(vAlterTbReq.nTagVal + 1, 1); + dataConverToStr(buf, vAlterTbReq.tagType, vAlterTbReq.pTagVal, vAlterTbReq.nTagVal, NULL); + } + + cJSON* colValue = cJSON_CreateString(buf); + cJSON_AddItemToObject(json, "colValue", colValue); + taosMemoryFree(buf); + } + + cJSON* isNullCJson = cJSON_CreateBool(isNull); + cJSON_AddItemToObject(json, "colValueNull", isNullCJson); break; } default: @@ -2171,18 +2299,18 @@ static char* processAlterTable(SMqMetaRsp* metaRsp) { } string = cJSON_PrintUnformatted(json); -_exit: + _exit: tDecoderClear(&decoder); return string; } -static char* processDropSTable(SMqMetaRsp* metaRsp) { - SDecoder decoder = {0}; - SVDropStbReq req = {0}; - char* string = NULL; +static char *processDropSTable(SMqMetaRsp *metaRsp){ + SDecoder decoder = {0}; + SVDropStbReq req = {0}; + char *string = NULL; // decode - void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead); tDecoderInit(&decoder, data, len); if (tDecodeSVDropStbReq(&decoder, &req) < 0) { @@ -2195,10 +2323,6 @@ static char* processDropSTable(SMqMetaRsp* metaRsp) { } cJSON* type = cJSON_CreateString("drop"); cJSON_AddItemToObject(json, "type", type); - char uid[32] = {0}; - sprintf(uid, "%" PRIi64, req.suid); - cJSON* id = cJSON_CreateString(uid); - cJSON_AddItemToObject(json, "id", id); cJSON* tableName = cJSON_CreateString(req.name); cJSON_AddItemToObject(json, "tableName", tableName); cJSON* tableType = cJSON_CreateString("super"); @@ -2206,18 +2330,18 @@ static char* processDropSTable(SMqMetaRsp* metaRsp) { string = cJSON_PrintUnformatted(json); -_exit: + _exit: tDecoderClear(&decoder); return string; } -static char* processDropTable(SMqMetaRsp* metaRsp) { - SDecoder decoder = {0}; - SVDropTbBatchReq req = {0}; - char* string = NULL; +static char *processDropTable(SMqMetaRsp *metaRsp){ + SDecoder decoder = {0}; + SVDropTbBatchReq req = {0}; + char *string = NULL; // decode - void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead)); int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead); tDecoderInit(&decoder, data, len); if (tDecodeSVDropTbBatchReq(&decoder, &req) < 0) { @@ -2230,67 +2354,71 @@ static char* processDropTable(SMqMetaRsp* metaRsp) { } cJSON* type = cJSON_CreateString("drop"); cJSON_AddItemToObject(json, "type", type); - // cJSON* uid = cJSON_CreateNumber(id); - // cJSON_AddItemToObject(json, "uid", uid); - // cJSON* tableType = cJSON_CreateString("normal"); - // cJSON_AddItemToObject(json, "tableType", tableType); +// cJSON* uid = cJSON_CreateNumber(id); +// cJSON_AddItemToObject(json, "uid", uid); +// cJSON* tableType = cJSON_CreateString("normal"); +// cJSON_AddItemToObject(json, "tableType", tableType); cJSON* tableNameList = cJSON_CreateArray(); for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { SVDropTbReq* pDropTbReq = req.pReqs + iReq; - cJSON* tableName = cJSON_CreateString(pDropTbReq->name); // todo + cJSON* tableName = cJSON_CreateString(pDropTbReq->name); cJSON_AddItemToArray(tableNameList, tableName); } cJSON_AddItemToObject(json, "tableNameList", tableNameList); string = cJSON_PrintUnformatted(json); -_exit: + _exit: tDecoderClear(&decoder); return string; } -char* tmq_get_json_meta(TAOS_RES* res) { +char *tmq_get_json_meta(TAOS_RES *res){ if (!TD_RES_TMQ_META(res)) { return NULL; } SMqMetaRspObj* pMetaRspObj = (SMqMetaRspObj*)res; - if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_STB) { + if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_STB){ return processCreateStb(&pMetaRspObj->metaRsp); - } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_STB) { - return processCreateStb(&pMetaRspObj->metaRsp); - } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_STB) { + }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_STB){ + return processAlterStb(&pMetaRspObj->metaRsp); + }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_STB){ return processDropSTable(&pMetaRspObj->metaRsp); - } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_TABLE) { + }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_CREATE_TABLE){ return processCreateTable(&pMetaRspObj->metaRsp); - } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_TABLE) { + }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_ALTER_TABLE){ return processAlterTable(&pMetaRspObj->metaRsp); - } else if (pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_TABLE) { + }else if(pMetaRspObj->metaRsp.resMsgType == TDMT_VND_DROP_TABLE){ return processDropTable(&pMetaRspObj->metaRsp); } return NULL; } -static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) { +void tmq_free_json_meta(char* jsonMeta){ + taosMemoryFreeClear(jsonMeta); +} + +static int32_t taosCreateStb(TAOS *taos, void *meta, int32_t metaLen){ SVCreateStbReq req = {0}; SDecoder coder; SMCreateStbReq pReq = {0}; - int32_t code = TSDB_CODE_SUCCESS; - SRequestObj* pRequest = NULL; + int32_t code = TSDB_CODE_SUCCESS; + SRequestObj* pRequest = NULL; code = buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest); if (code != TSDB_CODE_SUCCESS) { goto end; } - if (!pRequest->pDb) { + if(!pRequest->pDb){ code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; } // decode and process req - void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); int32_t len = metaLen - sizeof(SMsgHead); tDecoderInit(&coder, data, len); if (tDecodeSVCreateStbReq(&coder, &req) < 0) { @@ -2299,29 +2427,30 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) { } // build create stable pReq.pColumns = taosArrayInit(req.schemaRow.nCols, sizeof(SField)); - for (int32_t i = 0; i < req.schemaRow.nCols; i++) { + for(int32_t i = 0; i < req.schemaRow.nCols; i++){ SSchema* pSchema = req.schemaRow.pSchema + i; - SField field = {.type = pSchema->type, .bytes = pSchema->bytes}; + SField field = {.type = pSchema->type, .bytes = pSchema->bytes}; strcpy(field.name, pSchema->name); taosArrayPush(pReq.pColumns, &field); } pReq.pTags = taosArrayInit(req.schemaTag.nCols, sizeof(SField)); - for (int32_t i = 0; i < req.schemaTag.nCols; i++) { + for(int32_t i = 0; i < req.schemaTag.nCols; i++){ SSchema* pSchema = req.schemaTag.pSchema + i; - SField field = {.type = pSchema->type, .bytes = pSchema->bytes}; + SField field = {.type = pSchema->type, .bytes = pSchema->bytes}; strcpy(field.name, pSchema->name); taosArrayPush(pReq.pTags, &field); } + pReq.colVer = req.schemaRow.version; pReq.tagVer = req.schemaTag.version; pReq.numOfColumns = req.schemaRow.nCols; pReq.numOfTags = req.schemaTag.nCols; pReq.commentLen = -1; pReq.suid = req.suid; - pReq.source = 1; + pReq.source = TD_REQ_FROM_TAOX; + pReq.igExists = true; STscObj* pTscObj = pRequest->pTscObj; - SName tableName; tNameExtractFullName(toName(pTscObj->acctId, pRequest->pDb, req.name, &tableName), pReq.name); @@ -2336,7 +2465,7 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) { } tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq); - SQuery pQuery = {0}; + SQuery pQuery = {0}; pQuery.execMode = QUERY_EXEC_MODE_RPC; pQuery.pCmdMsg = &pCmdMsg; pQuery.msgType = pQuery.pCmdMsg->msgType; @@ -2346,18 +2475,18 @@ static int32_t taosCreateStb(TAOS* taos, void* meta, int32_t metaLen) { code = pRequest->code; taosMemoryFree(pCmdMsg.pMsg); -end: + end: destroyRequest(pRequest); tFreeSMCreateStbReq(&pReq); tDecoderClear(&coder); return code; } -static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { +static int32_t taosDropStb(TAOS *taos, void *meta, int32_t metaLen){ SVDropStbReq req = {0}; SDecoder coder; SMDropStbReq pReq = {0}; - int32_t code = TSDB_CODE_SUCCESS; + int32_t code = TSDB_CODE_SUCCESS; SRequestObj* pRequest = NULL; code = buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest); @@ -2365,12 +2494,12 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { goto end; } - if (!pRequest->pDb) { + if(!pRequest->pDb){ code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; } // decode and process req - void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); int32_t len = metaLen - sizeof(SMsgHead); tDecoderInit(&coder, data, len); if (tDecodeSVDropStbReq(&coder, &req) < 0) { @@ -2380,11 +2509,10 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { // build drop stable pReq.igNotExists = true; - pReq.source = 1; + pReq.source = TD_REQ_FROM_TAOX; pReq.suid = req.suid; STscObj* pTscObj = pRequest->pTscObj; - SName tableName; tNameExtractFullName(toName(pTscObj->acctId, pRequest->pDb, req.name, &tableName), pReq.name); @@ -2399,7 +2527,7 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { } tSerializeSMDropStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq); - SQuery pQuery = {0}; + SQuery pQuery = {0}; pQuery.execMode = QUERY_EXEC_MODE_RPC; pQuery.pCmdMsg = &pCmdMsg; pQuery.msgType = pQuery.pCmdMsg->msgType; @@ -2409,7 +2537,7 @@ static int32_t taosDropStb(TAOS* taos, void* meta, int32_t metaLen) { code = pRequest->code; taosMemoryFree(pCmdMsg.pMsg); -end: + end: destroyRequest(pRequest); tDecoderClear(&coder); return code; @@ -2422,7 +2550,7 @@ typedef struct SVgroupCreateTableBatch { } SVgroupCreateTableBatch; static void destroyCreateTbReqBatch(void* data) { - SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*)data; + SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*) data; taosArrayDestroy(pTbBatch->req.pArray); } @@ -2439,12 +2567,13 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) { goto end; } - if (!pRequest->pDb) { + + if(!pRequest->pDb){ code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; } // decode and process req - void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); int32_t len = metaLen - sizeof(SMsgHead); tDecoderInit(&coder, data, len); if (tDecodeSVCreateTbBatchReq(&coder, &req) < 0) { @@ -2469,15 +2598,15 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) { taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch); SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, - .requestId = pRequest->requestId, - .requestObjRefId = pRequest->self, - .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; // loop to create table for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { pCreateReq = req.pReqs + iReq; SVgroupInfo pInfo = {0}; - SName pName; + SName pName; toName(pTscObj->acctId, pRequest->pDb, pCreateReq->name, &pName); code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo); if (code != TSDB_CODE_SUCCESS) { @@ -2509,7 +2638,7 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) { pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE; pQuery->msgType = TDMT_VND_CREATE_TABLE; pQuery->stableQuery = false; - pQuery->pRoot = nodesMakeNode(QUERY_NODE_CREATE_TABLE_STMT); + pQuery->pRoot = nodesMakeNode(QUERY_NODE_CREATE_TABLE_STMT); code = rewriteToVnodeModifyOpStmt(pQuery, pBufArray); if (code != TSDB_CODE_SUCCESS) { @@ -2517,10 +2646,10 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) { } launchQueryImpl(pRequest, pQuery, false, NULL); - pQuery = NULL; // no need to free in the end - code = pRequest->code; + pQuery = NULL; // no need to free in the end + code = pRequest->code; -end: + end: taosHashCleanup(pVgroupHashmap); destroyRequest(pRequest); tDecoderClear(&coder); @@ -2552,12 +2681,12 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { goto end; } - if (!pRequest->pDb) { + if(!pRequest->pDb){ code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; } // decode and process req - void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); int32_t len = metaLen - sizeof(SMsgHead); tDecoderInit(&coder, data, len); if (tDecodeSVDropTbBatchReq(&coder, &req) < 0) { @@ -2582,15 +2711,16 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch); SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, - .requestId = pRequest->requestId, - .requestObjRefId = pRequest->self, - .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; // loop to create table for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { pDropReq = req.pReqs + iReq; + pDropReq->igNotExists = true; SVgroupInfo pInfo = {0}; - SName pName; + SName pName; toName(pTscObj->acctId, pRequest->pDb, pDropReq->name, &pName); code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo); if (code != TSDB_CODE_SUCCESS) { @@ -2620,7 +2750,7 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE; pQuery->msgType = TDMT_VND_DROP_TABLE; pQuery->stableQuery = false; - pQuery->pRoot = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT); + pQuery->pRoot = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT); code = rewriteToVnodeModifyOpStmt(pQuery, pBufArray); if (code != TSDB_CODE_SUCCESS) { @@ -2628,10 +2758,10 @@ static int32_t taosDropTable(TAOS* taos, void* meta, int32_t metaLen) { } launchQueryImpl(pRequest, pQuery, false, NULL); - pQuery = NULL; // no need to free in the end - code = pRequest->code; + pQuery = NULL; // no need to free in the end + code = pRequest->code; -end: + end: taosHashCleanup(pVgroupHashmap); destroyRequest(pRequest); tDecoderClear(&coder); @@ -2639,26 +2769,27 @@ end: return code; } -static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { - SVAlterTbReq req = {0}; - SDecoder coder = {0}; - int32_t code = TSDB_CODE_SUCCESS; - SRequestObj* pRequest = NULL; - SQuery* pQuery = NULL; - SArray* pArray = NULL; - SVgDataBlocks* pVgData = NULL; +static int32_t taosAlterTable(TAOS *taos, void *meta, int32_t metaLen){ + SVAlterTbReq req = {0}; + SDecoder coder = {0}; + int32_t code = TSDB_CODE_SUCCESS; + SRequestObj *pRequest = NULL; + SQuery *pQuery = NULL; + SArray *pArray = NULL; + SVgDataBlocks *pVgData = NULL; + + code = buildRequest(*(int64_t*) taos, "", 0, NULL, false, &pRequest); - code = buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest); if (code != TSDB_CODE_SUCCESS) { goto end; } - if (!pRequest->pDb) { + if(!pRequest->pDb){ code = TSDB_CODE_PAR_DB_NOT_SPECIFIED; goto end; } // decode and process req - void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); + void* data = POINTER_SHIFT(meta, sizeof(SMsgHead)); int32_t len = metaLen - sizeof(SMsgHead); tDecoderInit(&coder, data, len); if (tDecodeSVAlterTbReq(&coder, &req) < 0) { @@ -2667,24 +2798,24 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { } // do not deal TSDB_ALTER_TABLE_UPDATE_OPTIONS - if (req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS) { + if(req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS){ goto end; } STscObj* pTscObj = pRequest->pTscObj; - SCatalog* pCatalog = NULL; + SCatalog *pCatalog = NULL; code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog); if (code != TSDB_CODE_SUCCESS) { goto end; } SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter, - .requestId = pRequest->requestId, - .requestObjRefId = pRequest->self, - .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self, + .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)}; SVgroupInfo pInfo = {0}; - SName pName = {0}; + SName pName = {0}; toName(pTscObj->acctId, pRequest->pDb, req.tbName, &pName); code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo); if (code != TSDB_CODE_SUCCESS) { @@ -2718,7 +2849,7 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE; pQuery->msgType = TDMT_VND_ALTER_TABLE; pQuery->stableQuery = false; - pQuery->pRoot = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT); + pQuery->pRoot = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT); code = rewriteToVnodeModifyOpStmt(pQuery, pArray); if (code != TSDB_CODE_SUCCESS) { @@ -2726,14 +2857,17 @@ static int32_t taosAlterTable(TAOS* taos, void* meta, int32_t metaLen) { } launchQueryImpl(pRequest, pQuery, false, NULL); - pQuery = NULL; // no need to free in the end + pQuery = NULL; // no need to free in the end pVgData = NULL; - pArray = NULL; - code = pRequest->code; + pArray = NULL; + code = pRequest->code; + if (code == TSDB_CODE_VND_TABLE_NOT_EXIST){ + code = 0; + } -end: + end: taosArrayDestroy(pArray); - if (pVgData) taosMemoryFreeClear(pVgData->pData); + if(pVgData) taosMemoryFreeClear(pVgData->pData); taosMemoryFreeClear(pVgData); destroyRequest(pRequest); tDecoderClear(&coder); @@ -2741,29 +2875,33 @@ end: return code; } -int32_t taos_write_raw_meta(TAOS* taos, tmq_raw_data* raw_meta) { +int32_t taos_write_raw_meta(TAOS *taos, tmq_raw_data *raw_meta){ if (!taos || !raw_meta) { return TSDB_CODE_INVALID_PARA; } - if (raw_meta->raw_meta_type == TDMT_VND_CREATE_STB) { + if(raw_meta->raw_meta_type == TDMT_VND_CREATE_STB) { return taosCreateStb(taos, raw_meta->raw_meta, raw_meta->raw_meta_len); - } else if (raw_meta->raw_meta_type == TDMT_VND_ALTER_STB) { + }else if(raw_meta->raw_meta_type == TDMT_VND_ALTER_STB){ return taosCreateStb(taos, raw_meta->raw_meta, raw_meta->raw_meta_len); - } else if (raw_meta->raw_meta_type == TDMT_VND_DROP_STB) { + }else if(raw_meta->raw_meta_type == TDMT_VND_DROP_STB){ return taosDropStb(taos, raw_meta->raw_meta, raw_meta->raw_meta_len); - } else if (raw_meta->raw_meta_type == TDMT_VND_CREATE_TABLE) { + }else if(raw_meta->raw_meta_type == TDMT_VND_CREATE_TABLE){ return taosCreateTable(taos, raw_meta->raw_meta, raw_meta->raw_meta_len); - } else if (raw_meta->raw_meta_type == TDMT_VND_ALTER_TABLE) { + }else if(raw_meta->raw_meta_type == TDMT_VND_ALTER_TABLE){ return taosAlterTable(taos, raw_meta->raw_meta, raw_meta->raw_meta_len); - } else if (raw_meta->raw_meta_type == TDMT_VND_DROP_TABLE) { + }else if(raw_meta->raw_meta_type == TDMT_VND_DROP_TABLE){ return taosDropTable(taos, raw_meta->raw_meta, raw_meta->raw_meta_len); } return TSDB_CODE_INVALID_PARA; } +void tmq_free_raw_meta(tmq_raw_data *rawMeta){ + taosMemoryFreeClear(rawMeta); +} + void tmq_commit_async(tmq_t* tmq, const TAOS_RES* msg, tmq_commit_cb* cb, void* param) { tmqCommitInner2(tmq, msg, 0, 1, cb, param); } -int32_t tmq_commit_sync(tmq_t* tmq, const TAOS_RES* msg) { return tmqCommitInner2(tmq, msg, 0, 0, NULL, NULL); } +int32_t tmq_commit_sync(tmq_t* tmq, const TAOS_RES* msg) { return tmqCommitInner2(tmq, msg, 0, 0, NULL, NULL); } \ No newline at end of file diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index eca8d49b6a..9ebfa78b80 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -4871,6 +4871,11 @@ int tEncodeSVCreateStbReq(SEncoder *pCoder, const SVCreateStbReq *pReq) { if (tEncodeSRSmaParam(pCoder, &pReq->rsmaParam) < 0) return -1; } + if (tEncodeI32(pCoder, pReq->alterOriDataLen) < 0) return -1; + if (pReq->alterOriDataLen > 0) { + if (tEncodeBinary(pCoder, pReq->alterOriData, pReq->alterOriDataLen) < 0) return -1; + } + tEndEncode(pCoder); return 0; } @@ -4887,6 +4892,11 @@ int tDecodeSVCreateStbReq(SDecoder *pCoder, SVCreateStbReq *pReq) { if (tDecodeSRSmaParam(pCoder, &pReq->rsmaParam) < 0) return -1; } + if (tDecodeI32(pCoder, &pReq->alterOriDataLen) < 0) return -1; + if (pReq->alterOriDataLen > 0) { + if (tDecodeBinary(pCoder, (uint8_t **)&pReq->alterOriData, NULL) < 0) return -1; + } + tEndDecode(pCoder); return 0; } @@ -4930,8 +4940,15 @@ int tEncodeSVCreateTbReq(SEncoder *pCoder, const SVCreateTbReq *pReq) { } if (pReq->type == TSDB_CHILD_TABLE) { + if (tEncodeCStr(pCoder, pReq->ctb.name) < 0) return -1; if (tEncodeI64(pCoder, pReq->ctb.suid) < 0) return -1; if (tEncodeTag(pCoder, (const STag *)pReq->ctb.pTag) < 0) return -1; + int32_t len = taosArrayGetSize(pReq->ctb.tagName); + if (tEncodeI32(pCoder, len) < 0) return -1; + for (int32_t i = 0; i < len; i++){ + char* name = taosArrayGet(pReq->ctb.tagName, i); + if (tEncodeCStr(pCoder, name) < 0) return -1; + } } else if (pReq->type == TSDB_NORMAL_TABLE) { if (tEncodeSSchemaWrapper(pCoder, &pReq->ntb.schemaRow) < 0) return -1; } else { @@ -4959,8 +4976,20 @@ int tDecodeSVCreateTbReq(SDecoder *pCoder, SVCreateTbReq *pReq) { } if (pReq->type == TSDB_CHILD_TABLE) { + if (tDecodeCStr(pCoder, &pReq->ctb.name) < 0) return -1; if (tDecodeI64(pCoder, &pReq->ctb.suid) < 0) return -1; if (tDecodeTag(pCoder, (STag **)&pReq->ctb.pTag) < 0) return -1; + int32_t len = 0; + if (tDecodeI32(pCoder, &len) < 0) return -1; + pReq->ctb.tagName = taosArrayInit(len, TSDB_COL_NAME_LEN); + if(pReq->ctb.tagName == NULL) return -1; + for (int32_t i = 0; i < len; i++){ + char name[TSDB_COL_NAME_LEN] = {0}; + char *tmp = NULL; + if (tDecodeCStr(pCoder, &tmp) < 0) return -1; + strcpy(name, tmp); + taosArrayPush(pReq->ctb.tagName, name); + } } else if (pReq->type == TSDB_NORMAL_TABLE) { if (tDecodeSSchemaWrapperEx(pCoder, &pReq->ntb.schemaRow) < 0) return -1; } else { @@ -5292,6 +5321,7 @@ int32_t tEncodeSVAlterTbReq(SEncoder *pEncoder, const SVAlterTbReq *pReq) { break; case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: if (tEncodeCStr(pEncoder, pReq->colName) < 0) return -1; + if (tEncodeI8(pEncoder, pReq->colModType) < 0) return -1; if (tEncodeI32v(pEncoder, pReq->colModBytes) < 0) return -1; break; case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: @@ -5301,6 +5331,7 @@ int32_t tEncodeSVAlterTbReq(SEncoder *pEncoder, const SVAlterTbReq *pReq) { case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: if (tEncodeCStr(pEncoder, pReq->tagName) < 0) return -1; if (tEncodeI8(pEncoder, pReq->isNull) < 0) return -1; + if (tEncodeI8(pEncoder, pReq->tagType) < 0) return -1; if (!pReq->isNull) { if (tEncodeBinary(pEncoder, pReq->pTagVal, pReq->nTagVal) < 0) return -1; } @@ -5340,6 +5371,7 @@ int32_t tDecodeSVAlterTbReq(SDecoder *pDecoder, SVAlterTbReq *pReq) { break; case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: if (tDecodeCStr(pDecoder, &pReq->colName) < 0) return -1; + if (tDecodeI8(pDecoder, &pReq->colModType) < 0) return -1; if (tDecodeI32v(pDecoder, &pReq->colModBytes) < 0) return -1; break; case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: @@ -5349,6 +5381,7 @@ int32_t tDecodeSVAlterTbReq(SDecoder *pDecoder, SVAlterTbReq *pReq) { case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: if (tDecodeCStr(pDecoder, &pReq->tagName) < 0) return -1; if (tDecodeI8(pDecoder, &pReq->isNull) < 0) return -1; + if (tDecodeI8(pDecoder, &pReq->tagType) < 0) return -1; if (!pReq->isNull) { if (tDecodeBinary(pDecoder, &pReq->pTagVal, &pReq->nTagVal) < 0) return -1; } diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index 1e57625028..bff33af5af 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -45,7 +45,7 @@ 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); -static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp); +static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp, void* alterOriData, int32_t alterOriDataLen); int32_t mndInitStb(SMnode *pMnode) { SSdbTable table = { @@ -409,7 +409,7 @@ static FORCE_INLINE int32_t schemaExColIdCompare(const void *colId, const void * return 0; } -static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) { +static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen, void* alterOriData, int32_t alterOriDataLen) { SEncoder encoder = {0}; int32_t contLen; SName name = {0}; @@ -422,6 +422,8 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt req.name = (char *)tNameGetTableName(&name); req.suid = pStb->uid; req.rollup = pStb->ast1Len > 0 ? 1 : 0; + req.alterOriData = alterOriData; + req.alterOriDataLen = alterOriDataLen; // todo req.schemaRow.nCols = pStb->numOfColumns; req.schemaRow.version = pStb->colVer; @@ -626,7 +628,7 @@ static int32_t mndSetCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj continue; } - void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen); + void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0); if (pReq == NULL) { sdbCancelFetch(pSdb, pIter); sdbRelease(pSdb, pVgroup); @@ -706,10 +708,10 @@ int32_t mndBuildStbFromReq(SMnode *pMnode, SStbObj *pDst, SMCreateStbReq *pCreat memcpy(pDst->db, pDb->name, TSDB_DB_FNAME_LEN); pDst->createdTime = taosGetTimestampMs(); pDst->updateTime = pDst->createdTime; - pDst->uid = (pCreate->source == 1) ? pCreate->suid : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN); + pDst->uid = (pCreate->source == TD_REQ_FROM_TAOX) ? pCreate->suid : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN); pDst->dbUid = pDb->uid; - pDst->tagVer = (pCreate->source != TD_REQ_FROM_APP) ? pCreate->tagVer : 1; - pDst->colVer = (pCreate->source != TD_REQ_FROM_APP) ? pCreate->colVer : 1; + pDst->tagVer = 1; + pDst->colVer = 1; pDst->smaVer = 1; pDst->nextColId = 1; pDst->maxdelay[0] = pCreate->delay1; @@ -849,6 +851,75 @@ static int32_t mndProcessTtlTimer(SRpcMsg *pReq) { return 0; } +static int32_t mndFindSuperTableTagIndex(const SStbObj *pStb, const char *tagName) { + for (int32_t tag = 0; tag < pStb->numOfTags; tag++) { + if (strcasecmp(pStb->pTags[tag].name, tagName) == 0) { + return tag; + } + } + + return -1; +} + +static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *colName) { + for (int32_t col = 0; col < pStb->numOfColumns; col++) { + if (strcasecmp(pStb->pColumns[col].name, colName) == 0) { + return col; + } + } + + return -1; +} + +static int32_t mndBuildStbFromAlter(SStbObj *pStb, SStbObj *pDst, SMCreateStbReq *createReq) { + taosRLockLatch(&pStb->lock); + memcpy(pDst, pStb, sizeof(SStbObj)); + taosRUnLockLatch(&pStb->lock); + + pDst->updateTime = taosGetTimestampMs(); + pDst->numOfColumns = createReq->numOfColumns; + pDst->numOfTags = createReq->numOfTags; + pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema)); + pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema)); + if (pDst->pColumns == NULL || pDst->pTags == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + for (int32_t i = 0; i < pDst->numOfColumns; ++i) { + SField *pField = taosArrayGet(createReq->pColumns, i); + SSchema *pSchema = &pDst->pColumns[i]; + pSchema->type = pField->type; + pSchema->bytes = pField->bytes; + pSchema->flags = pField->flags; + memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN); + int32_t cIndex = mndFindSuperTableColumnIndex(pStb, pField->name); + if (cIndex >= 0){ + pSchema->colId = pStb->pColumns[cIndex].colId; + }else{ + pSchema->colId = pDst->nextColId++; + } + } + + for (int32_t i = 0; i < pDst->numOfTags; ++i) { + SField *pField = taosArrayGet(createReq->pTags, i); + SSchema *pSchema = &pDst->pTags[i]; + pSchema->type = pField->type; + pSchema->bytes = pField->bytes; + memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN); + int32_t cIndex = mndFindSuperTableTagIndex(pStb, pField->name); + if (cIndex >= 0){ + pSchema->colId = pStb->pTags[cIndex].colId; + }else{ + pSchema->colId = pDst->nextColId++; + } + + } + pDst->tagVer = createReq->tagVer; + pDst->colVer = createReq->colVer; + return TSDB_CODE_SUCCESS; +} + static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) { SMnode *pMnode = pReq->info.node; int32_t code = -1; @@ -881,9 +952,9 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) { terrno = TSDB_CODE_MND_STABLE_UID_NOT_MATCH; goto _OVER; } else if (createReq.tagVer > 0 || createReq.colVer > 0) { - int32_t tagDelta = pStb->tagVer - createReq.tagVer; - int32_t colDelta = pStb->colVer - createReq.colVer; - int32_t verDelta = tagDelta + verDelta; + int32_t tagDelta = createReq.tagVer - pStb->tagVer; + int32_t colDelta = createReq.colVer - pStb->colVer; + int32_t verDelta = tagDelta + colDelta; mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d", createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer); if (tagDelta <= 0 && colDelta <= 0) { @@ -910,6 +981,10 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) { } } else if (terrno != TSDB_CODE_MND_STB_NOT_EXIST) { goto _OVER; + } else if (createReq.source == TD_REQ_FROM_TAOX && (createReq.tagVer != 1 || createReq.colVer != 1)){ + mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name); + code = 0; + goto _OVER; } pDb = mndAcquireDbByStb(pMnode, createReq.name); @@ -934,7 +1009,16 @@ static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) { if (isAlter) { bool needRsp = false; - code = mndAlterStbImp(pMnode, pReq, pDb, pStb, needRsp); + SStbObj pDst = {0}; + if (mndBuildStbFromAlter(pStb, &pDst, &createReq) != 0) { + taosMemoryFreeClear(pDst.pTags); + taosMemoryFreeClear(pDst.pColumns); + goto _OVER; + } + + code = mndAlterStbImp(pMnode, pReq, pDb, &pDst, needRsp, NULL, 0); + taosMemoryFreeClear(pDst.pTags); + taosMemoryFreeClear(pDst.pColumns); } else { code = mndCreateStb(pMnode, pReq, &createReq, pDb); } @@ -972,26 +1056,6 @@ static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) { return 0; } -static int32_t mndFindSuperTableTagIndex(const SStbObj *pStb, const char *tagName) { - for (int32_t tag = 0; tag < pStb->numOfTags; tag++) { - if (strcasecmp(pStb->pTags[tag].name, tagName) == 0) { - return tag; - } - } - - return -1; -} - -static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *colName) { - for (int32_t col = 0; col < pStb->numOfColumns; col++) { - if (strcasecmp(pStb->pColumns[col].name, colName) == 0) { - return col; - } - } - - return -1; -} - static int32_t mndAllocStbSchemas(const SStbObj *pOld, SStbObj *pNew) { pNew->pTags = taosMemoryCalloc(pNew->numOfTags, sizeof(SSchema)); pNew->pColumns = taosMemoryCalloc(pNew->numOfColumns, sizeof(SSchema)); @@ -1315,7 +1379,7 @@ static int32_t mndSetAlterStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj * return 0; } -static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) { +static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb, void* alterOriData, int32_t alterOriDataLen) { SSdb *pSdb = pMnode->pSdb; SVgObj *pVgroup = NULL; void *pIter = NULL; @@ -1329,7 +1393,7 @@ static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj continue; } - void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen); + void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen); if (pReq == NULL) { sdbCancelFetch(pSdb, pIter); sdbRelease(pSdb, pVgroup); @@ -1542,7 +1606,7 @@ static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, i return 0; } -static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp) { +static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp, void* alterOriData, int32_t alterOriDataLen) { int32_t code = -1; STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq); if (pTrans == NULL) goto _OVER; @@ -1559,7 +1623,7 @@ static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbOb if (mndSetAlterStbRedoLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER; if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER; - if (mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb) != 0) goto _OVER; + if (mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen) != 0) goto _OVER; if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER; code = 0; @@ -1620,7 +1684,7 @@ static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *p } if (code != 0) goto _OVER; - code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp); + code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen); _OVER: taosMemoryFreeClear(stbObj.pTags); @@ -1785,8 +1849,8 @@ static int32_t mndProcessDropStbReq(SRpcMsg *pReq) { } } - if (dropReq.source != TD_REQ_FROM_APP && pStb->uid != dropReq.suid) { - terrno = TSDB_CODE_MND_STB_NOT_EXIST; + if (dropReq.source == TD_REQ_FROM_TAOX && pStb->uid != dropReq.suid) { + code = 0; goto _OVER; } diff --git a/source/dnode/mnode/sdb/inc/sdb.h b/source/dnode/mnode/sdb/inc/sdb.h index 1294f0cff3..be56d901de 100644 --- a/source/dnode/mnode/sdb/inc/sdb.h +++ b/source/dnode/mnode/sdb/inc/sdb.h @@ -163,6 +163,7 @@ typedef struct SSdbRow { ESdbType type; ESdbStatus status; int32_t refCount; + int64_t forAlign; char pObj[]; } SSdbRow; diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index daf7ccb26a..29c10c40c2 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -1125,6 +1125,11 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) { skmDbKey.uid = pME->uid; skmDbKey.sver = pSW->version; + // if receive tmq meta message is: create stable1 then delete stable1 then create stable1 with multi vgroups + if (tdbTbGet(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), NULL, NULL) == 0) { + return rcode; + } + // encode schema int32_t ret = 0; tEncodeSize(tEncodeSSchemaWrapper, pSW, vLen, ret); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index df80a4c218..6f4a04ee09 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -401,7 +401,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) { tqDebug("fetch meta msg, ver:%" PRId64 ", type:%d", pHead->version, pHead->msgType); SMqMetaRsp metaRsp = {0}; /*metaRsp.reqOffset = pReq->reqOffset.version;*/ - /*metaRsp.rspOffset = fetchVer;*/ + metaRsp.rspOffset = fetchVer; /*metaRsp.rspOffsetNew.version = fetchVer;*/ tqOffsetResetToLog(&metaRsp.reqOffsetNew, pReq->reqOffset.version); tqOffsetResetToLog(&metaRsp.rspOffsetNew, fetchVer); diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c index dbbb2b2661..25615f8d5c 100644 --- a/source/dnode/vnode/src/tq/tqSink.c +++ b/source/dnode/vnode/src/tq/tqSink.c @@ -50,6 +50,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo SVCreateTbReq createTbReq = {0}; createTbReq.name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId); + createTbReq.ctb.name = strdup(stbFullName); createTbReq.flags = 0; createTbReq.type = TSDB_CHILD_TABLE; createTbReq.ctb.suid = suid; diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index d25ae817c7..cca212a4e4 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -494,8 +494,6 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pR taosArrayPush(rsp.pArray, &cRsp); } - tDecoderClear(&decoder); - tqUpdateTbUidList(pVnode->pTq, tbUids, true); tdUpdateTbUidList(pVnode->pSma, pStore); tdUidStoreFree(pStore); @@ -512,9 +510,12 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pR } tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen); tEncodeSVCreateTbBatchRsp(&encoder, &rsp); - tEncoderClear(&encoder); _exit: + for (int32_t iReq = 0; iReq < req.nReqs; iReq++) { + pCreateReq = req.pReqs + iReq; + taosArrayDestroy(pCreateReq->ctb.tagName); + } taosArrayDestroy(rsp.pArray); taosArrayDestroy(tbUids); tDecoderClear(&decoder); @@ -611,7 +612,7 @@ static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pRe // process if (metaAlterTable(pVnode->pMeta, version, &vAlterTbReq, &vMetaRsp) < 0) { - vAlterTbRsp.code = TSDB_CODE_INVALID_MSG; + vAlterTbRsp.code = terrno; tDecoderClear(&dc); rcode = -1; goto _exit; @@ -795,6 +796,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) { pRsp->code = TSDB_CODE_INVALID_MSG; tDecoderClear(&decoder); + taosArrayDestroy(createTbReq.ctb.tagName); goto _exit; } @@ -802,6 +804,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) { submitBlkRsp.code = terrno; tDecoderClear(&decoder); + taosArrayDestroy(createTbReq.ctb.tagName); goto _exit; } } @@ -822,6 +825,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid"); #endif tDecoderClear(&decoder); + taosArrayDestroy(createTbReq.ctb.tagName); } else { submitBlkRsp.tblFName = taosMemoryMalloc(TSDB_TABLE_FNAME_LEN); sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname); diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 4b8eed57d7..95b4fdcd6e 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -307,7 +307,6 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SNode* pTagIndexCond = (SNode*)pListInfo->pTagIndexCond; if (pScanNode->tableType == TSDB_SUPER_TABLE) { if (pTagIndexCond) { - ///<<<<<<< HEAD SIndexMetaArg metaArg = { .metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid}; @@ -315,20 +314,9 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, SIdxFltStatus status = SFLT_NOT_INDEX; code = doFilterTag(pTagIndexCond, &metaArg, res, &status); if (code != 0 || status == SFLT_NOT_INDEX) { - code = TSDB_CODE_INDEX_REBUILDING; - } - //======= - // SArray* res = taosArrayInit(8, sizeof(uint64_t)); - // // code = doFilterTag(pTagIndexCond, &metaArg, res); - // code = TSDB_CODE_INDEX_REBUILDING; - //>>>>>>> dvv - if (code == TSDB_CODE_INDEX_REBUILDING) { + qError("failed to get tableIds from index, reason:%s, suid:%" PRIu64, tstrerror(code), tableUid); +// code = TSDB_CODE_INDEX_REBUILDING; code = vnodeGetAllTableList(pVnode, tableUid, pListInfo->pTableList); - } else if (code != TSDB_CODE_SUCCESS) { - qError("failed to get tableIds, reason:%s, suid:%" PRIu64, tstrerror(code), tableUid); - taosArrayDestroy(res); - terrno = code; - return code; } else { qDebug("success to get tableIds, size:%d, suid:%" PRIu64, (int)taosArrayGetSize(res), tableUid); } @@ -347,25 +335,25 @@ int32_t getTableList(void* metaHandle, void* pVnode, SScanPhysiNode* pScanNode, terrno = code; return code; } - - if (pTagCond) { - int32_t i = 0; - while (i < taosArrayGetSize(pListInfo->pTableList)) { - STableKeyInfo* info = taosArrayGet(pListInfo->pTableList, i); - bool isOk = isTableOk(info, pTagCond, metaHandle); - if (terrno) return terrno; - if (!isOk) { - taosArrayRemove(pListInfo->pTableList, i); - continue; - } - i++; - } - } } else { // Create one table group. STableKeyInfo info = {.lastKey = 0, .uid = tableUid, .groupId = 0}; taosArrayPush(pListInfo->pTableList, &info); } + if (pTagCond) { + int32_t i = 0; + while (i < taosArrayGetSize(pListInfo->pTableList)) { + STableKeyInfo* info = taosArrayGet(pListInfo->pTableList, i); + bool isOk = isTableOk(info, pTagCond, metaHandle); + if (terrno) return terrno; + if (!isOk) { + taosArrayRemove(pListInfo->pTableList, i); + continue; + } + i++; + } + } + pListInfo->pGroupList = taosArrayInit(4, POINTER_BYTES); if (pListInfo->pGroupList == NULL) { return TSDB_CODE_OUT_OF_MEMORY; diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c index f986d24a7e..005a7f919a 100644 --- a/source/libs/parser/src/parInsert.c +++ b/source/libs/parser/src/parInsert.c @@ -71,6 +71,7 @@ typedef struct SInsertParseContext { SVnodeModifOpStmt* pOutput; SStmtCallback* pStmtCb; SParseMetaCache* pMetaCache; + char sTableName[TSDB_TABLE_NAME_LEN]; } SInsertParseContext; typedef struct SInsertParseSyntaxCxt { @@ -734,11 +735,13 @@ static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo* return TSDB_CODE_SUCCESS; } -static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid) { +static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid, const char* sname, SArray* tagName) { pTbReq->type = TD_CHILD_TABLE; pTbReq->name = strdup(tname); pTbReq->ctb.suid = suid; + if(sname) pTbReq->ctb.name = strdup(sname); pTbReq->ctb.pTag = (uint8_t*)pTag; + pTbReq->ctb.tagName = taosArrayDup(tagName); pTbReq->commentLen = -1; return; @@ -758,6 +761,7 @@ static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16 return TSDB_CODE_SUCCESS; } +// strcpy(val->colName, pSchema->name); val->cid = pSchema->colId; val->type = pSchema->type; @@ -936,6 +940,7 @@ static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16 static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) { int32_t code = TSDB_CODE_SUCCESS; SArray* pTagVals = taosArrayInit(pCxt->tags.numOfBound, sizeof(STagVal)); + SArray* tagName = taosArrayInit(8, TSDB_COL_NAME_LEN); SToken sToken; bool isParseBindParam = false; bool isJson = false; @@ -965,6 +970,10 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint taosMemoryFree(tmpTokenBuf); goto end; } + + if (!isNullStr(&sToken)) { + taosArrayPush(tagName, pTagSchema->name); + } if (pTagSchema->type == TSDB_DATA_TYPE_JSON) { if (sToken.n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) { code = buildSyntaxErrMsg(&pCxt->msg, "json string too long than 4095", sToken.z); @@ -1004,7 +1013,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint goto end; } - buildCreateTbReq(&pCxt->createTblReq, tName, pTag, pCxt->pTableMeta->suid); + buildCreateTbReq(&pCxt->createTblReq, tName, pTag, pCxt->pTableMeta->suid, pCxt->sTableName, tagName); end: for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) { @@ -1014,6 +1023,7 @@ end: } } taosArrayDestroy(pTagVals); + taosArrayDestroy(tagName); return code; } @@ -1089,6 +1099,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tb createSName(&sname, &sToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg); char dbFName[TSDB_DB_FNAME_LEN]; tNameGetFullDbName(&sname, dbFName); + strcpy(pCxt->sTableName, sname.tname); CHECK_CODE(getSTableMeta(pCxt, &sname, dbFName)); if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) { @@ -1325,15 +1336,10 @@ static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STa return TSDB_CODE_SUCCESS; } -void destroyCreateSubTbReq(SVCreateTbReq* pReq) { - taosMemoryFreeClear(pReq->name); - taosMemoryFreeClear(pReq->ctb.pTag); -} - static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) { taosMemoryFreeClear(pCxt->pTableMeta); destroyBoundColumnInfo(&pCxt->tags); - destroyCreateSubTbReq(&pCxt->createTblReq); + tdDestroySVCreateTbReq(&pCxt->createTblReq); } static void destroySubTableHashElem(void* p) { taosMemoryFree(*(STableMeta**)p); } @@ -1479,7 +1485,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { } memcpy(tags, &pCxt->tags, sizeof(pCxt->tags)); (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl, - pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj); + pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj, pCxt->sTableName); memset(&pCxt->tags, 0, sizeof(pCxt->tags)); pCxt->pVgroupsHashObj = NULL; @@ -1508,6 +1514,7 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache .pSql = (char*)pContext->pSql, .msg = {.buf = pContext->pMsg, .len = pContext->msgLen}, .pTableMeta = NULL, + .createTblReq = {0}, .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK), .pTableNameHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK), .pDbFNameHashObj = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK), @@ -1788,7 +1795,7 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash return TSDB_CODE_SUCCESS; } -int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tName, TAOS_MULTI_BIND* bind, +int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const char* sTableName, char* tName, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen) { STableDataBlocks* pDataBlock = (STableDataBlocks*)pBlock; SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen}; @@ -1802,6 +1809,11 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN return buildInvalidOperationMsg(&pBuf, "out of memory"); } + SArray* tagName = taosArrayInit(8, TSDB_COL_NAME_LEN); + if (!tagName) { + return buildInvalidOperationMsg(&pBuf, "out of memory"); + } + int32_t code = TSDB_CODE_SUCCESS; SSchema* pSchema = getTableTagSchema(pDataBlock->pTableMeta); @@ -1818,6 +1830,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN if (IS_VAR_DATA_TYPE(pTagSchema->type)) { colLen = bind[c].length[0]; } + taosArrayPush(tagName, pTagSchema->name); if (pTagSchema->type == TSDB_DATA_TYPE_JSON) { if (colLen > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) { code = buildSyntaxErrMsg(&pBuf, "json string too long than 4095", bind[c].buffer); @@ -1834,6 +1847,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN } } else { STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type}; +// strcpy(val.colName, pTagSchema->name); if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) { val.pData = (uint8_t*)bind[c].buffer; val.nData = colLen; @@ -1870,9 +1884,9 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tN } SVCreateTbReq tbReq = {0}; - buildCreateTbReq(&tbReq, tName, pTag, suid); + buildCreateTbReq(&tbReq, tName, pTag, suid, sTableName, tagName); code = buildCreateTbMsg(pDataBlock, &tbReq); - destroyCreateSubTbReq(&tbReq); + tdDestroySVCreateTbReq(&tbReq); end: for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) { @@ -1882,6 +1896,7 @@ end: } } taosArrayDestroy(pTagArray); + taosArrayDestroy(tagName); return code; } @@ -2136,7 +2151,7 @@ typedef struct SmlExecHandle { static void smlDestroyTableHandle(void* pHandle) { SmlExecTableHandle* handle = (SmlExecTableHandle*)pHandle; destroyBoundColumnInfo(&handle->tags); - destroyCreateSubTbReq(&handle->createTblReq); + tdDestroySVCreateTbReq(&handle->createTblReq); } static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SSchema* pSchema) { @@ -2222,18 +2237,24 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS * @param msg * @return int32_t */ -static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SMsgBuf* msg) { +static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SArray** tagName, SMsgBuf* msg) { SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal)); if (!pTagArray) { return TSDB_CODE_TSC_OUT_OF_MEMORY; } + *tagName = taosArrayInit(8, TSDB_COL_NAME_LEN); + if (!*tagName) { + return TSDB_CODE_TSC_OUT_OF_MEMORY; + } int32_t code = TSDB_CODE_SUCCESS; for (int i = 0; i < tags->numOfBound; ++i) { SSchema* pTagSchema = &pSchema[tags->boundColumns[i]]; SSmlKv* kv = taosArrayGetP(cols, i); + taosArrayPush(*tagName, pTagSchema->name); STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type}; +// strcpy(val.colName, pTagSchema->name); if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) { val.pData = (uint8_t*)kv->value; val.nData = kv->length; @@ -2277,7 +2298,7 @@ end: } int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta, - char* tableName, char* msgBuf, int16_t msgBufLen) { + char* tableName, const char* sTableName, int32_t sTableNameLen, char* msgBuf, int16_t msgBufLen) { SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen}; SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle; @@ -2290,12 +2311,19 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols return ret; } STag* pTag = NULL; - ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, &pTag, &pBuf); + SArray* tagName = NULL; + ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, &pTag, &tagName, &pBuf); if (ret != TSDB_CODE_SUCCESS) { + taosArrayDestroy(tagName); return ret; } - buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, pTag, pTableMeta->suid); + buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, pTag, pTableMeta->suid, NULL, tagName); + taosArrayDestroy(tagName); + + smlHandle->tableExecHandle.createTblReq.ctb.name = taosMemoryMalloc(sTableNameLen + 1); + memcpy(smlHandle->tableExecHandle.createTblReq.ctb.name, sTableName, sTableNameLen); + smlHandle->tableExecHandle.createTblReq.ctb.name[sTableNameLen] = 0; STableDataBlocks* pDataBlock = NULL; ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid), diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index d12f1cfe7c..0542b4eec7 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -3754,6 +3754,9 @@ static int32_t buildCreateStbReq(STranslateContext* pCxt, SCreateTableStmt* pStm pReq->delay2 = pStmt->pOptions->maxDelay2; pReq->watermark1 = pStmt->pOptions->watermark1; pReq->watermark2 = pStmt->pOptions->watermark2; + pReq->colVer = 1; + pReq->tagVer = 1; + pReq->source = TD_REQ_FROM_APP; columnDefNodeToField(pStmt->pCols, &pReq->pColumns); columnDefNodeToField(pStmt->pTags, &pReq->pTags); pReq->numOfColumns = LIST_LENGTH(pStmt->pCols); @@ -5337,6 +5340,8 @@ static void destroyCreateTbReqBatch(void* data) { taosMemoryFreeClear(pTableReq->ntb.schemaRow.pSchema); } else if (pTableReq->type == TSDB_CHILD_TABLE) { taosMemoryFreeClear(pTableReq->ctb.pTag); + taosMemoryFreeClear(pTableReq->ctb.name); + taosArrayDestroy(pTableReq->ctb.tagName); } } @@ -5408,11 +5413,11 @@ static int32_t rewriteCreateTable(STranslateContext* pCxt, SQuery* pQuery) { } static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, SCreateSubTableClause* pStmt, - const STag* pTag, uint64_t suid, SVgroupInfo* pVgInfo) { - // char dbFName[TSDB_DB_FNAME_LEN] = {0}; - // SName name = {.type = TSDB_DB_NAME_T, .acctId = acctId}; - // strcpy(name.dbname, pStmt->dbName); - // tNameGetFullDbName(&name, dbFName); + const STag* pTag, uint64_t suid, const char* sTableNmae, SVgroupInfo* pVgInfo, SArray* tagName) { +// char dbFName[TSDB_DB_FNAME_LEN] = {0}; +// SName name = {.type = TSDB_DB_NAME_T, .acctId = acctId}; +// strcpy(name.dbname, pStmt->dbName); +// tNameGetFullDbName(&name, dbFName); struct SVCreateTbReq req = {0}; req.type = TD_CHILD_TABLE; @@ -5425,7 +5430,9 @@ static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, S req.commentLen = -1; } req.ctb.suid = suid; + req.ctb.name = strdup(sTableNmae); req.ctb.pTag = (uint8_t*)pTag; + req.ctb.tagName = taosArrayDup(tagName); if (pStmt->ignoreExists) { req.flags |= TD_CREATE_IF_NOT_EXISTS; } @@ -5517,6 +5524,7 @@ static int32_t buildNormalTagVal(STranslateContext* pCxt, SSchema* pTagSchema, S if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) { void* nodeVal = nodesGetValueFromNode(pVal); STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type}; +// strcpy(val.colName, pTagSchema->name); if (IS_VAR_DATA_TYPE(pTagSchema->type)) { val.pData = varDataVal(nodeVal); val.nData = varDataLen(nodeVal); @@ -5529,7 +5537,7 @@ static int32_t buildNormalTagVal(STranslateContext* pCxt, SSchema* pTagSchema, S } static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta, - STag** ppTag) { + STag** ppTag, SArray* tagName) { int32_t numOfTags = getNumOfTags(pSuperTableMeta); if (LIST_LENGTH(pStmt->pValsOfTags) != LIST_LENGTH(pStmt->pSpecificTags) || numOfTags < LIST_LENGTH(pStmt->pValsOfTags)) { @@ -5560,8 +5568,10 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla if (pSchema->type == TSDB_DATA_TYPE_JSON) { isJson = true; code = buildJsonTagVal(pCxt, pSchema, pVal, pTagArray, ppTag); + taosArrayPush(tagName, pCol->colName); } else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL) { code = buildNormalTagVal(pCxt, pSchema, pVal, pTagArray); + taosArrayPush(tagName, pCol->colName); } } if (TSDB_CODE_SUCCESS == code) { @@ -5582,7 +5592,7 @@ static int32_t buildKVRowForBindTags(STranslateContext* pCxt, SCreateSubTableCla } static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClause* pStmt, STableMeta* pSuperTableMeta, - STag** ppTag) { + STag** ppTag, SArray* tagName) { if (getNumOfTags(pSuperTableMeta) != LIST_LENGTH(pStmt->pValsOfTags)) { return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_TAGS_NOT_MATCHED); } @@ -5607,9 +5617,11 @@ static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClau if (pTagSchema->type == TSDB_DATA_TYPE_JSON) { isJson = true; code = buildJsonTagVal(pCxt, pTagSchema, pVal, pTagArray, ppTag); + taosArrayPush(tagName, pTagSchema->name); } else if (pVal->node.resType.type != TSDB_DATA_TYPE_NULL && !pVal->isNull) { char* tmpVal = nodesGetValueFromNode(pVal); STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type}; +// strcpy(val.colName, pTagSchema->name); if (IS_VAR_DATA_TYPE(pTagSchema->type)) { val.pData = varDataVal(tmpVal); val.nData = varDataLen(tmpVal); @@ -5617,6 +5629,7 @@ static int32_t buildKVRowForAllTags(STranslateContext* pCxt, SCreateSubTableClau memcpy(&val.i64, tmpVal, pTagSchema->bytes); } taosArrayPush(pTagArray, &val); + taosArrayPush(tagName, pTagSchema->name); } } if (TSDB_CODE_SUCCESS == code) { @@ -5652,12 +5665,13 @@ static int32_t rewriteCreateSubTable(STranslateContext* pCxt, SCreateSubTableCla } STag* pTag = NULL; + SArray* tagName = taosArrayInit(8, TSDB_COL_NAME_LEN); if (TSDB_CODE_SUCCESS == code) { if (NULL != pStmt->pSpecificTags) { - code = buildKVRowForBindTags(pCxt, pStmt, pSuperTableMeta, &pTag); + code = buildKVRowForBindTags(pCxt, pStmt, pSuperTableMeta, &pTag, tagName); } else { - code = buildKVRowForAllTags(pCxt, pStmt, pSuperTableMeta, &pTag); + code = buildKVRowForAllTags(pCxt, pStmt, pSuperTableMeta, &pTag, tagName); } } @@ -5666,9 +5680,10 @@ static int32_t rewriteCreateSubTable(STranslateContext* pCxt, SCreateSubTableCla code = getTableHashVgroup(pCxt, pStmt->dbName, pStmt->tableName, &info); } if (TSDB_CODE_SUCCESS == code) { - addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt, pTag, pSuperTableMeta->uid, &info); + addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt, pTag, pSuperTableMeta->uid, pStmt->useTableName, &info, tagName); } + taosArrayDestroy(tagName); taosMemoryFreeClear(pSuperTableMeta); return code; } @@ -5883,9 +5898,8 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS return pCxt->errCode; } - pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type); + pReq->tagType = targetDt.type; if (targetDt.type == TSDB_DATA_TYPE_JSON) { - pReq->isNull = 0; if (pStmt->pVal->literal && strlen(pStmt->pVal->literal) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) { return buildSyntaxErrMsg(&pCxt->msgBuf, "json string too long than 4095", pStmt->pVal->literal); @@ -5913,6 +5927,7 @@ static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pS pReq->pTagVal = (uint8_t*)pTag; pStmt->pVal->datum.p = (char*)pTag; // for free } else { + pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type); pReq->nTagVal = pStmt->pVal->node.resType.bytes; pReq->pTagVal = nodesGetValueFromNode(pStmt->pVal); @@ -5966,7 +5981,7 @@ static int32_t buildDropColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, static int32_t buildUpdateColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta, SVAlterTbReq* pReq) { pReq->colModBytes = calcTypeBytes(pStmt->dataType); - + pReq->colModType = pStmt->dataType.type; SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName); if (NULL == pSchema) { return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName); diff --git a/source/libs/parser/src/parUtil.c b/source/libs/parser/src/parUtil.c index 3484fa61c7..51f59c9362 100644 --- a/source/libs/parser/src/parUtil.c +++ b/source/libs/parser/src/parUtil.c @@ -389,6 +389,7 @@ int32_t parseJsontoTagData(const char* json, SArray* pTagVals, STag** ppTag, voi continue; } STagVal val = {0}; +// strcpy(val.colName, colName); val.pKey = jsonKey; taosHashPut(keyHash, jsonKey, keyLen, &keyLen, CHAR_BYTES); // add key to hash to remove dumplicate, value is useless