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

Merge branch '3.0' into fix/TD-31891-remove-void-monitor2

This commit is contained in:
dongming chen 2024-09-23 14:07:57 +08:00 committed by GitHub
parent 160440a8e4 5639fd0baf
commit 062e446495
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
36 changed files with 7778 additions and 7188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/zh/14-reference/02-tools/08-taos-cli.md
View File

@ -82,7 +82,7 @@ taos -h h1.taos.com -s "use db; show tables;"
## 配置文件
也可以通过配置文件中的参数设置来控制 TDengine CLI 的行为。可用配置参数请参考[客户端配置](../taosc)
也可以通过配置文件中的参数设置来控制 TDengine CLI 的行为。可用配置参数请参考[客户端配置](../../components/taosc)
## TDengine CLI 小技巧

5
include/common/tmsg.h
View File

@ -462,6 +462,7 @@ typedef enum ENodeType {
typedef struct {
int32_t vgId;
uint8_t option; // 0x0 REQ_OPT_TBNAME, 0x01 REQ_OPT_TBUID
const char* dbFName;
const char* tbName;
} SBuildTableInput;
@ -1459,6 +1460,7 @@ typedef struct {
int32_t walFsyncPeriod;
int16_t hashPrefix;
int16_t hashSuffix;
int8_t hashMethod;
int8_t walLevel;
int8_t precision;
int8_t compression;
@ -2044,10 +2046,13 @@ typedef struct {
int32_t tSerializeSAlterVnodeHashRangeReq(void* buf, int32_t bufLen, SAlterVnodeHashRangeReq* pReq);
int32_t tDeserializeSAlterVnodeHashRangeReq(void* buf, int32_t bufLen, SAlterVnodeHashRangeReq* pReq);
#define REQ_OPT_TBNAME 0x0
#define REQ_OPT_TBUID 0x01
typedef struct {
SMsgHead header;
char dbFName[TSDB_DB_FNAME_LEN];
char tbName[TSDB_TABLE_NAME_LEN];
uint8_t option;
} STableInfoReq;
int32_t tSerializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);

1
include/common/tmsgdef.h
View File

@ -309,6 +309,7 @@
TD_DEF_MSG_TYPE(TDMT_VND_ARB_HEARTBEAT, "vnode-arb-hb", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_ARB_CHECK_SYNC, "vnode-arb-check-sync", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_FETCH_TTL_EXPIRED_TBS, "vnode-fetch-ttl-expired-tbs", NULL, NULL)
TD_DEF_MSG_TYPE(TDMT_VND_TABLE_NAME, "vnode-table-name", NULL, NULL)
TD_CLOSE_MSG_SEG(TDMT_VND_MSG)
TD_NEW_MSG_SEG(TDMT_SCH_MSG) // 3<<8

3
include/libs/catalog/catalog.h
View File

@ -96,6 +96,7 @@ typedef struct SCatalogReq {
SArray* pView; // element is STablesReq
SArray* pTableTSMAs; // element is STablesReq
SArray* pTSMAs; // element is STablesReq
SArray* pTableName; // element is STablesReq
bool qNodeRequired; // valid qnode
bool dNodeRequired; // valid dnode
bool svrVerRequired;
@ -417,6 +418,8 @@ int32_t catalogGetTsma(SCatalog* pCtg, SRequestConnInfo* pConn, const SName* pTs
int32_t catalogAsyncUpdateDbTsmaVersion(SCatalog* pCtg, int32_t tsmaVersion, const char* dbFName, int64_t dbId);
int32_t ctgHashValueComp(void const* lp, void const* rp);
/**
* Destroy catalog and relase all resources
*/

2
include/libs/nodes/cmdnodes.h
View File

@ -240,6 +240,7 @@ typedef struct SDropTableStmt {
ENodeType type;
SNodeList* pTables;
bool withTsma;
bool withOpt;
} SDropTableStmt;
typedef struct SDropSuperTableStmt {
@ -247,6 +248,7 @@ typedef struct SDropSuperTableStmt {
char dbName[TSDB_DB_NAME_LEN];
char tableName[TSDB_TABLE_NAME_LEN];
bool ignoreNotExists;
bool withOpt;
} SDropSuperTableStmt;
typedef struct SAlterTableStmt {

1
include/libs/qcom/query.h
View File

@ -330,6 +330,7 @@ int32_t getAsofJoinReverseOp(EOperatorType op);
int32_t queryCreateCTableMetaFromMsg(STableMetaRsp* msg, SCTableMeta* pMeta);
int32_t queryCreateTableMetaFromMsg(STableMetaRsp* msg, bool isSuperTable, STableMeta** pMeta);
int32_t queryCreateTableMetaExFromMsg(STableMetaRsp* msg, bool isSuperTable, STableMeta** pMeta);
char* jobTaskStatusStr(int32_t status);
SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name);

1
source/common/src/systable.c
View File

@ -159,6 +159,7 @@ static const SSysDbTableSchema userStbsSchema[] = {
{.name = "watermark", .bytes = 64 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "max_delay", .bytes = 64 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "rollup", .bytes = 128 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "uid", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT, .sysInfo = false},
};
static const SSysDbTableSchema streamSchema[] = {

12
source/common/src/tmsg.c
View File

@ -4912,6 +4912,7 @@ int32_t tSerializeSDbCfgRspImpl(SEncoder *encoder, const SDbCfgRsp *pRsp) {
TAOS_CHECK_RETURN(tEncodeI32(encoder, pRsp->s3ChunkSize));
TAOS_CHECK_RETURN(tEncodeI32(encoder, pRsp->s3KeepLocal));
TAOS_CHECK_RETURN(tEncodeI8(encoder, pRsp->s3Compact));
TAOS_CHECK_RETURN(tEncodeI8(encoder, pRsp->hashMethod));
return 0;
}
@ -5003,6 +5004,11 @@ int32_t tDeserializeSDbCfgRspImpl(SDecoder *decoder, SDbCfgRsp *pRsp) {
TAOS_CHECK_RETURN(tDecodeI32(decoder, &pRsp->s3KeepLocal));
TAOS_CHECK_RETURN(tDecodeI8(decoder, &pRsp->s3Compact));
}
if (!tDecodeIsEnd(decoder)) {
TAOS_CHECK_RETURN(tDecodeI8(decoder, &pRsp->hashMethod));
} else {
pRsp->hashMethod = 1; // default value
}
return 0;
}
@ -5720,6 +5726,7 @@ int32_t tSerializeSTableInfoReq(void *buf, int32_t bufLen, STableInfoReq *pReq)
TAOS_CHECK_EXIT(tStartEncode(&encoder));
TAOS_CHECK_EXIT(tEncodeCStr(&encoder, pReq->dbFName));
TAOS_CHECK_EXIT(tEncodeCStr(&encoder, pReq->tbName));
TAOS_CHECK_EXIT(tEncodeU8(&encoder, pReq->option));
tEndEncode(&encoder);
_exit:
@ -5754,6 +5761,11 @@ int32_t tDeserializeSTableInfoReq(void *buf, int32_t bufLen, STableInfoReq *pReq
TAOS_CHECK_EXIT(tStartDecode(&decoder));
TAOS_CHECK_EXIT(tDecodeCStrTo(&decoder, pReq->dbFName));
TAOS_CHECK_EXIT(tDecodeCStrTo(&decoder, pReq->tbName));
if (!tDecodeIsEnd(&decoder)) {
TAOS_CHECK_EXIT(tDecodeU8(&decoder, &pReq->option));
} else {
pReq->option = 0;
}
tEndDecode(&decoder);
_exit:

2
source/common/test/msgTypeTable.ini
View File

@ -382,6 +382,8 @@ TDMT_VND_ARB_CHECK_SYNC = 613
TDMT_VND_ARB_CHECK_SYNC_RSP = 614
TDMT_VND_FETCH_TTL_EXPIRED_TBS = 615
TDMT_VND_FETCH_TTL_EXPIRED_TBS_RSP = 616
TDMT_VND_TABLE_NAME = 617
TDMT_VND_TABLE_NAME_RSP = 618
TDMT_SCH_QUERY = 769
TDMT_SCH_QUERY_RSP = 770
TDMT_SCH_MERGE_QUERY = 771

1
source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
View File

@ -992,6 +992,7 @@ SArray *vmGetMsgHandles() {
if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_INDEX, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_QUERY_COMPACT_PROGRESS, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_KILL_COMPACT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_NAME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

1
source/dnode/mnode/impl/src/mndDb.c
View File

@ -1329,6 +1329,7 @@ static void mndDumpDbCfgInfo(SDbCfgRsp *cfgRsp, SDbObj *pDb) {
cfgRsp->walFsyncPeriod = pDb->cfg.walFsyncPeriod;
cfgRsp->hashPrefix = pDb->cfg.hashPrefix;
cfgRsp->hashSuffix = pDb->cfg.hashSuffix;
cfgRsp->hashMethod = pDb->cfg.hashMethod;
cfgRsp->walLevel = pDb->cfg.walLevel;
cfgRsp->precision = pDb->cfg.precision;
cfgRsp->compression = pDb->cfg.compression;

5
source/dnode/mnode/impl/src/mndStb.c
View File

@ -3626,6 +3626,11 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
if (pColInfo) {
RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->uid), false), pStb, &lino, _ERROR);
}
numOfRows++;
sdbRelease(pSdb, pStb);
}

2
source/dnode/vnode/inc/vnode.h
View File

@ -125,7 +125,7 @@ int32_t metaGetTableTags(void *pVnode, uint64_t suid, SArray *uidList);
int32_t metaGetTableTagsByUids(void *pVnode, int64_t suid, SArray *uidList);
int32_t metaReadNext(SMetaReader *pReader);
const void *metaGetTableTagVal(const void *tag, int16_t type, STagVal *tagVal);
int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
int32_t metaGetTableNameByUid(void *pVnode, uint64_t uid, char *tbName);
int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName);
int metaGetTableUidByName(void *pVnode, char *tbName, uint64_t *uid);

36
source/dnode/vnode/src/vnd/vnodeQuery.c
View File

@ -55,6 +55,7 @@ int32_t vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
SMetaReader mer1 = {0};
SMetaReader mer2 = {0};
char tableFName[TSDB_TABLE_FNAME_LEN];
bool reqTbUid = false;
SRpcMsg rpcMsg = {0};
int32_t code = 0;
int32_t rspLen = 0;
@ -68,20 +69,33 @@ int32_t vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
goto _exit4;
}
if (infoReq.option == REQ_OPT_TBUID) reqTbUid = true;
metaRsp.dbId = pVnode->config.dbId;
(void)strcpy(metaRsp.tbName, infoReq.tbName);
(void)memcpy(metaRsp.dbFName, infoReq.dbFName, sizeof(metaRsp.dbFName));
(void)sprintf(tableFName, "%s.%s", infoReq.dbFName, infoReq.tbName);
code = vnodeValidateTableHash(pVnode, tableFName);
if (code) {
goto _exit4;
if (!reqTbUid) {
TAOS_UNUSED(sprintf(tableFName, "%s.%s", infoReq.dbFName, infoReq.tbName));
code = vnodeValidateTableHash(pVnode, tableFName);
if (code) {
goto _exit4;
}
}
// query meta
metaReaderDoInit(&mer1, pVnode->pMeta, META_READER_LOCK);
if (metaGetTableEntryByName(&mer1, infoReq.tbName) < 0) {
if (reqTbUid) {
errno = 0;
uint64_t tbUid = taosStr2UInt64(infoReq.tbName, NULL, 10);
if (errno == ERANGE || tbUid == 0) {
code = TSDB_CODE_TDB_TABLE_NOT_EXIST;
goto _exit3;
}
char tbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
TAOS_CHECK_GOTO(metaGetTableNameByUid(pVnode, tbUid, tbName), NULL, _exit3);
tstrncpy(metaRsp.tbName, varDataVal(tbName), TSDB_TABLE_NAME_LEN);
TAOS_CHECK_GOTO(metaGetTableEntryByName(&mer1, varDataVal(tbName)), NULL, _exit3);
} else if (metaGetTableEntryByName(&mer1, infoReq.tbName) < 0) {
code = terrno;
goto _exit3;
}
@ -107,7 +121,7 @@ int32_t vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {
schema = mer1.me.ntbEntry.schemaRow;
} else {
vError("vnodeGetTableMeta get invalid table type:%d", mer1.me.type);
return TSDB_CODE_APP_ERROR;
goto _exit3;
}
metaRsp.numOfTags = schemaTag.nCols;
@ -175,7 +189,7 @@ _exit4:
rpcMsg.msgType = pMsg->msgType;
if (code) {
qError("get table %s meta failed cause of %s", infoReq.tbName, tstrerror(code));
qError("get table %s meta with %" PRIu8 " failed cause of %s", infoReq.tbName, infoReq.option, tstrerror(code));
}
if (direct) {
@ -400,6 +414,12 @@ int32_t vnodeGetBatchMeta(SVnode *pVnode, SRpcMsg *pMsg) {
qWarn("vnodeGetBatchMeta failed, msgType:%d", req->msgType);
}
break;
case TDMT_VND_TABLE_NAME:
// error code has been set into reqMsg, no need to handle it here.
if (TSDB_CODE_SUCCESS != vnodeGetTableMeta(pVnode, &reqMsg, false)) {
qWarn("vnodeGetBatchName failed, msgType:%d", req->msgType);
}
break;
case TDMT_VND_TABLE_CFG:
// error code has been set into reqMsg, no need to handle it here.
if (TSDB_CODE_SUCCESS != vnodeGetTableCfg(pVnode, &reqMsg, false)) {

3
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -786,7 +786,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg);
if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||
pMsg->msgType == TDMT_VND_BATCH_META) &&
pMsg->msgType == TDMT_VND_BATCH_META || pMsg->msgType == TDMT_VND_TABLE_NAME) &&
!syncIsReadyForRead(pVnode->sync)) {
vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
return 0;
@ -807,6 +807,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
case TDMT_SCH_QUERY_HEARTBEAT:
return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);
case TDMT_VND_TABLE_META:
case TDMT_VND_TABLE_NAME:
return vnodeGetTableMeta(pVnode, pMsg, true);
case TDMT_VND_TABLE_CFG:
return vnodeGetTableCfg(pVnode, pMsg, true);

15
source/libs/catalog/inc/catalogInt.h
View File

@ -132,6 +132,7 @@ typedef enum {
CTG_TASK_GET_VIEW,
CTG_TASK_GET_TB_TSMA,
CTG_TASK_GET_TSMA,
CTG_TASK_GET_TB_NAME,
} CTG_TASK_TYPE;
typedef enum {
@ -205,6 +206,14 @@ typedef struct SCtgTbMetasCtx {
SArray* pFetchs;
} SCtgTbMetasCtx;
typedef struct SCtgTbNamesCtx {
int32_t fetchNum;
SRWLatch lock;
SArray* pNames;
SArray* pResList;
SArray* pFetchs;
} SCtgTbNamesCtx;
typedef struct SCtgTbIndexCtx {
SName* pName;
} SCtgTbIndexCtx;
@ -421,6 +430,7 @@ typedef struct SCtgJob {
int32_t viewNum;
int32_t tbTsmaNum;
int32_t tsmaNum; // currently, only 1 is possible
int32_t tbNameNum;
} SCtgJob;
typedef struct SCtgMsgCtx {
@ -797,7 +807,8 @@ typedef struct SCtgCacheItemInfo {
#define CTG_IS_BATCH_TASK(_taskType) \
((CTG_TASK_GET_TB_META_BATCH == (_taskType)) || (CTG_TASK_GET_TB_HASH_BATCH == (_taskType)) || \
(CTG_TASK_GET_VIEW == (_taskType)) || (CTG_TASK_GET_TB_TSMA == (_taskType)))
(CTG_TASK_GET_VIEW == (_taskType)) || (CTG_TASK_GET_TB_TSMA == (_taskType)) || \
(CTG_TASK_GET_TB_NAME == (_taskType)))
#define CTG_GET_TASK_MSGCTX(_task, _id) \
(CTG_IS_BATCH_TASK((_task)->type) ? taosArrayGet((_task)->msgCtxs, (_id)) : &(_task)->msgCtx)
@ -996,6 +1007,8 @@ int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq
int32_t ctgGetTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
int32_t ctgGetTbMetasFromCache(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbMetasCtx* ctx, int32_t dbIdx,
int32_t* fetchIdx, int32_t baseResIdx, SArray* pList);
int32_t ctgGetTbNamesFromCache(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgTbNamesCtx* ctx, int32_t dbIdx,
int32_t* fetchIdx, int32_t baseResIdx, SArray* pList);
int32_t ctgCloneDbCfgInfo(void* pSrc, SDbCfgInfo** ppDst);
int32_t ctgOpUpdateVgroup(SCtgCacheOperation* action);

308
source/libs/catalog/src/ctgAsync.c
View File

@ -588,6 +588,35 @@ int32_t ctgInitGetTSMATask(SCtgJob* pJob, int32_t taskId, void* param) {
return TSDB_CODE_SUCCESS;
}
static int32_t ctgInitGetTbNamesTask(SCtgJob* pJob, int32_t taskId, void* param) {
SCtgTask task = {0};
task.type = CTG_TASK_GET_TB_NAME;
task.taskId = taskId;
task.pJob = pJob;
SCtgTbNamesCtx* pTaskCtx = taosMemoryCalloc(1, sizeof(SCtgTbNamesCtx));
if (NULL == pTaskCtx) {
CTG_ERR_RET(terrno);
}
task.taskCtx = pTaskCtx;
taosInitRWLatch(&pTaskCtx->lock);
pTaskCtx->pNames = param;
pTaskCtx->pResList = taosArrayInit(pJob->tbNameNum, sizeof(SMetaRes));
if (NULL == pTaskCtx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbNameNum,
(int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true);
CTG_ERR_RET(terrno);
}
if (NULL == taosArrayPush(pJob->pTasks, &task)) {
ctgFreeTask(&task, true);
CTG_ERR_RET(terrno);
}
return TSDB_CODE_SUCCESS;
}
int32_t ctgHandleForceUpdateView(SCatalog* pCtg, const SCatalogReq* pReq) {
int32_t viewNum = taosArrayGetSize(pReq->pView);
for (int32_t i = 0; i < viewNum; ++i) {
@ -822,9 +851,10 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const
int32_t viewNum = (int32_t)ctgGetTablesReqNum(pReq->pView);
int32_t tbTsmaNum = (int32_t)taosArrayGetSize(pReq->pTableTSMAs);
int32_t tsmaNum = (int32_t)taosArrayGetSize(pReq->pTSMAs);
int32_t tbNameNum = (int32_t)ctgGetTablesReqNum(pReq->pTableName);
int32_t taskNum = tbMetaNum + dbVgNum + udfNum + tbHashNum + qnodeNum + dnodeNum + svrVerNum + dbCfgNum + indexNum +
userNum + dbInfoNum + tbIndexNum + tbCfgNum + tbTagNum + viewNum + tbTsmaNum;
userNum + dbInfoNum + tbIndexNum + tbCfgNum + tbTagNum + viewNum + tbTsmaNum + tbNameNum;
*job = taosMemoryCalloc(1, sizeof(SCtgJob));
if (NULL == *job) {
@ -859,6 +889,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const
pJob->viewNum = viewNum;
pJob->tbTsmaNum = tbTsmaNum;
pJob->tsmaNum = tsmaNum;
pJob->tbNameNum = tbNameNum;
#if CTG_BATCH_FETCH
pJob->pBatchs =
@ -991,6 +1022,9 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const
if (tsmaNum > 0) {
CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_TSMA, pReq->pTSMAs, NULL));
}
if (tbNameNum > 0) {
CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_TB_NAME, pReq->pTableName, NULL));
}
if (qnodeNum) {
CTG_ERR_JRET(ctgInitTask(pJob, CTG_TASK_GET_QNODE, NULL, NULL));
}
@ -1059,6 +1093,18 @@ int32_t ctgDumpTbMetasRes(SCtgTask* pTask) {
return TSDB_CODE_SUCCESS;
}
static int32_t ctgDumpTbNamesRes(SCtgTask* pTask) {
if (pTask->subTask) {
return TSDB_CODE_SUCCESS;
}
SCtgJob* pJob = pTask->pJob;
pJob->jobRes.pTableMeta = pTask->res;
return TSDB_CODE_SUCCESS;
}
int32_t ctgDumpDbVgRes(SCtgTask* pTask) {
if (pTask->subTask) {
return TSDB_CODE_SUCCESS;
@ -1858,6 +1904,201 @@ _return:
CTG_RET(code);
}
static int32_t ctgHandleGetTbNamesRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBuf* pMsg, int32_t rspCode) {
int32_t code = 0;
SCtgDBCache* dbCache = NULL;
SCtgTask* pTask = tReq->pTask;
SCatalog* pCtg = pTask->pJob->pCtg;
SRequestConnInfo* pConn = &pTask->pJob->conn;
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, tReq->msgIdx);
SCtgTbNamesCtx* ctx = (SCtgTbNamesCtx*)pTask->taskCtx;
bool taskDone = false;
bool lock = false;
if (NULL == pMsgCtx) {
ctgError("fail to get task msgCtx, taskType:%d", pTask->type);
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
SCtgFetch* pFetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
if (NULL == pFetch) {
ctgError("fail to get the %dth fetch, fetchNum:%d", tReq->msgIdx, (int32_t)taosArrayGetSize(ctx->pFetchs));
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
int32_t flag = pFetch->flag;
int32_t* vgId = &pFetch->vgId;
SName* pName = NULL;
CTG_ERR_JRET(ctgGetFetchName(ctx->pNames, pFetch, &pName));
if (reqType == TDMT_VND_TABLE_NAME) {
taosWLockLatch(&ctx->lock);
lock = true;
}
CTG_ERR_JRET(ctgProcessRspMsg(pMsgCtx->out, reqType, pMsg->pData, pMsg->len, rspCode, pMsgCtx->target));
switch (reqType) {
case TDMT_MND_USE_DB: {
SUseDbOutput* pOut = (SUseDbOutput*)pMsgCtx->out;
CTG_ERR_RET(ctgMakeVgArray(pOut->dbVgroup));
SArray* pVgArray = NULL;
TSWAP(pVgArray, pOut->dbVgroup->vgArray);
int32_t vgSize = taosArrayGetSize(pVgArray);
if (0 == vgSize) {
taosArrayDestroy(pVgArray);
ctgTaskError("no vgroup got, dbName:%s", pName->dbname);
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
for (int32_t i = 0; i < vgSize; ++i) {
SVgroupInfo* vgInfo = TARRAY_GET_ELEM(pVgArray, i);
if (NULL == vgInfo) {
taosArrayDestroy(pVgArray);
ctgTaskError("fail to get the %dth vgInfo, vgSize:%d", i, vgSize);
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
ctgTaskDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d, vgId:%d",
tNameGetTableName(pName), flag, vgInfo->vgId);
// *vgId = vgInfo->vgId;
if (i > 0) atomic_add_fetch_32(&ctx->fetchNum, 1);
code = ctgGetTbMetaFromVnode(pCtg, pConn, pName, vgInfo, NULL, tReq);
if (code) {
taosArrayDestroy(pVgArray);
CTG_ERR_JRET(code);
}
}
taosArrayDestroy(pVgArray);
return TSDB_CODE_SUCCESS;
}
case TDMT_MND_TABLE_META: {
STableMetaOutput* pOut = (STableMetaOutput*)pMsgCtx->out;
if (CTG_IS_META_NULL(pOut->metaType)) {
if (CTG_FLAG_IS_STB(flag)) {
char dbFName[TSDB_DB_FNAME_LEN] = {0};
(void)tNameGetFullDbName(pName, dbFName);
CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache));
if (NULL != dbCache) {
SVgroupInfo vgInfo = {0};
CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, &pConn->mgmtEps, dbCache->vgCache.vgInfo, pName, &vgInfo));
ctgTaskDebug("will refresh tbmeta, supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pName), flag);
*vgId = vgInfo.vgId;
CTG_ERR_JRET(ctgGetTbMetaFromVnode(pCtg, pConn, pName, &vgInfo, NULL, tReq));
ctgReleaseVgInfoToCache(pCtg, dbCache);
dbCache = NULL;
} else {
SBuildUseDBInput input = {0};
tstrncpy(input.db, dbFName, tListLen(input.db));
input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
CTG_ERR_JRET(ctgGetDBVgInfoFromMnode(pCtg, pConn, &input, NULL, tReq));
}
return TSDB_CODE_SUCCESS;
}
ctgTaskError("no tbmeta got, tbName:%s", tNameGetTableName(pName));
(void)ctgRemoveTbMetaFromCache(pCtg, pName, false); // cache update not fatal error
CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
}
if (pMsgCtx->lastOut) {
TSWAP(pMsgCtx->out, pMsgCtx->lastOut);
STableMetaOutput* pLastOut = (STableMetaOutput*)pMsgCtx->out;
TSWAP(pLastOut->tbMeta, pOut->tbMeta);
}
break;
}
case TDMT_VND_TABLE_NAME: {
STableMetaOutput* pOut = (STableMetaOutput*)pMsgCtx->out;
if (CTG_IS_META_NULL(pOut->metaType)) {
ctgTaskError("no tbmeta got, tbName:%s", tNameGetTableName(pName));
CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
}
break;
}
default:
ctgTaskError("invalid reqType %d", reqType);
CTG_ERR_JRET(TSDB_CODE_INVALID_MSG);
}
STableMetaOutput* pOut = (STableMetaOutput*)pMsgCtx->out;
if (CTG_IS_META_BOTH(pOut->metaType)) {
TAOS_MEMCPY(pOut->tbMeta, &pOut->ctbMeta, sizeof(pOut->ctbMeta));
}
SMetaRes* pRes = taosArrayGet(ctx->pResList, pFetch->resIdx);
if (NULL == pRes) {
ctgTaskError("fail to get the %dth res in pResList, resNum:%d", pFetch->resIdx,
(int32_t)taosArrayGetSize(ctx->pResList));
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
if (!pRes->pRes) {
pRes->code = 0;
pRes->pRes = pOut->tbMeta;
pOut->tbMeta = NULL;
} else {
taosMemoryFreeClear(pOut->tbMeta);
}
if (0 == atomic_sub_fetch_32(&ctx->fetchNum, 1)) {
TSWAP(pTask->res, ctx->pResList);
taskDone = true;
}
_return:
if (dbCache) {
ctgReleaseVgInfoToCache(pCtg, dbCache);
}
if (code) {
SMetaRes* pRes = taosArrayGet(ctx->pResList, pFetch->resIdx);
if (NULL == pRes) {
ctgTaskError("fail to get the %dth res in pResList, resNum:%d", pFetch->resIdx,
(int32_t)taosArrayGetSize(ctx->pResList));
} else {
if (0 == atomic_sub_fetch_32(&ctx->fetchNum, 1)) {
TSWAP(pTask->res, ctx->pResList);
taskDone = true;
}
if (TDMT_VND_TABLE_NAME == reqType) {
if (!pRes->pRes && (0 == pRes->code)) pRes->code = code;
} else {
pRes->pRes = NULL;
pRes->code = code;
}
if (taskDone == true) {
ctgTaskError("Get table %d.%s.%s meta failed with error %s", pName->acctId, pName->dbname, pName->tname,
tstrerror(code));
}
}
}
if (pTask->res && taskDone) {
int32_t newCode = ctgHandleTaskEnd(pTask, code);
if (newCode && TSDB_CODE_SUCCESS == code) {
code = newCode;
}
}
if (lock) {
taosWUnLockLatch(&ctx->lock);
}
CTG_RET(code);
}
int32_t ctgHandleGetDbVgRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBuf* pMsg, int32_t rspCode) {
int32_t code = 0;
SCtgTask* pTask = tReq->pTask;
@ -3730,6 +3971,70 @@ int32_t ctgLaunchGetTSMATask(SCtgTask* pTask) {
return TSDB_CODE_SUCCESS;
}
static int32_t ctgLaunchGetTbNamesTask(SCtgTask* pTask) {
SCatalog* pCtg = pTask->pJob->pCtg;
SRequestConnInfo* pConn = &pTask->pJob->conn;
SCtgTbNamesCtx* pCtx = (SCtgTbNamesCtx*)pTask->taskCtx;
SCtgJob* pJob = pTask->pJob;
SName* pName = NULL;
int32_t dbNum = taosArrayGetSize(pCtx->pNames);
int32_t fetchIdx = 0;
int32_t baseResIdx = 0;
for (int32_t i = 0; i < dbNum; ++i) {
STablesReq* pReq = TARRAY_GET_ELEM(pCtx->pNames, i);
if (NULL == pReq) {
ctgError("fail to get the %dth STablesReq, num:%d", i, dbNum);
CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
}
ctgDebug("start to check tbname metas in db %s, tbNum %ld", pReq->dbFName, taosArrayGetSize(pReq->pTables));
CTG_ERR_RET(ctgGetTbNamesFromCache(pCtg, pConn, pCtx, i, &fetchIdx, baseResIdx, pReq->pTables));
baseResIdx += taosArrayGetSize(pReq->pTables);
}
pCtx->fetchNum = taosArrayGetSize(pCtx->pFetchs);
if (pCtx->fetchNum <= 0) {
TSWAP(pTask->res, pCtx->pResList);
CTG_ERR_RET(ctgHandleTaskEnd(pTask, 0));
return TSDB_CODE_SUCCESS;
}
pTask->msgCtxs = taosArrayInit_s(sizeof(SCtgMsgCtx), pCtx->fetchNum);
if (NULL == pTask->msgCtxs) {
ctgError("taosArrayInit_s %d SCtgMsgCtx %d failed", pCtx->fetchNum, (int32_t)sizeof(SCtgMsgCtx));
CTG_ERR_RET(terrno);
}
for (int32_t i = 0; i < pCtx->fetchNum; ++i) {
SCtgFetch* pFetch = taosArrayGet(pCtx->pFetchs, i);
if (NULL == pFetch) {
ctgError("fail to get the %dth fetch in pCtx->pFetchs, fetchNum:%d", i, pCtx->fetchNum);
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
CTG_ERR_RET(ctgGetFetchName(pCtx->pNames, pFetch, &pName));
SCtgMsgCtx* pMsgCtx = CTG_GET_TASK_MSGCTX(pTask, i);
if (NULL == pMsgCtx) {
ctgError("fail to get the %dth pMsgCtx", i);
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
if (NULL == pMsgCtx->pBatchs) {
pMsgCtx->pBatchs = pJob->pBatchs;
}
SCtgTaskReq tReq;
tReq.pTask = pTask;
tReq.msgIdx = pFetch->fetchIdx;
CTG_ERR_RET(ctgAsyncRefreshTbMeta(&tReq, pFetch->flag, pName, &pFetch->vgId));
}
return TSDB_CODE_SUCCESS;
}
int32_t ctgDumpTbTSMARes(SCtgTask* pTask) {
if (pTask->subTask) {
return TSDB_CODE_SUCCESS;
@ -3876,6 +4181,7 @@ SCtgAsyncFps gCtgAsyncFps[] = {
{ctgInitGetViewsTask, ctgLaunchGetViewsTask, ctgHandleGetViewsRsp, ctgDumpViewsRes, NULL, NULL},
{ctgInitGetTbTSMATask, ctgLaunchGetTbTSMATask, ctgHandleGetTbTSMARsp, ctgDumpTbTSMARes, NULL, NULL},
{ctgInitGetTSMATask, ctgLaunchGetTSMATask, ctgHandleGetTSMARsp, ctgDumpTSMARes, NULL, NULL},
{ctgInitGetTbNamesTask, ctgLaunchGetTbNamesTask, ctgHandleGetTbNamesRsp, ctgDumpTbNamesRes, NULL, NULL},
};
int32_t ctgMakeAsyncRes(SCtgJob* pJob) {

33
source/libs/catalog/src/ctgCache.c
View File

@ -3691,6 +3691,39 @@ _return:
CTG_RET(code);
}
int32_t ctgGetTbNamesFromCache(SCatalog *pCtg, SRequestConnInfo *pConn, SCtgTbNamesCtx *ctx, int32_t dbIdx,
int32_t *fetchIdx, int32_t baseResIdx, SArray *pList) {
int32_t tbNum = taosArrayGetSize(pList);
char dbFName[TSDB_DB_FNAME_LEN] = {0};
int32_t flag = CTG_FLAG_UNKNOWN_STB;
int32_t code = TSDB_CODE_SUCCESS;
uint64_t lastSuid = 0;
STableMeta *lastTableMeta = NULL;
SName *pName = taosArrayGet(pList, 0);
if (NULL == pName) {
ctgError("fail to get the 0th SName from tableList, tableNum:%d", (int32_t)taosArrayGetSize(pList));
CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
}
if (IS_SYS_DBNAME(pName->dbname)) {
CTG_FLAG_SET_SYS_DB(flag);
TAOS_STRCPY(dbFName, pName->dbname);
} else {
(void)tNameGetFullDbName(pName, dbFName);
}
ctgDebug("db %s not in cache", dbFName);
for (int32_t i = 0; i < tbNum; ++i) {
CTG_ERR_JRET(ctgAddFetch(&ctx->pFetchs, dbIdx, i, fetchIdx, baseResIdx + i, flag));
if (NULL == taosArrayPush(ctx->pResList, &(SMetaData){0})) {
CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
}
}
_return:
return code;
}
int32_t ctgGetTbHashVgroupFromCache(SCatalog *pCtg, const SName *pTableName, SVgroupInfo **pVgroup) {
if (IS_SYS_DBNAME(pTableName->dbname)) {
ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);

45
source/libs/catalog/src/ctgRemote.c
View File

@ -297,6 +297,27 @@ 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_NAME: {
if (TSDB_CODE_SUCCESS != rspCode) {
if (CTG_TABLE_NOT_EXIST(rspCode)) {
SET_META_TYPE_NULL(((STableMetaOutput*)out)->metaType);
qDebug("tablemeta not exist in vnode, tbFName:%s", target);
return TSDB_CODE_SUCCESS;
}
qError("error rsp for table meta 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 tablemeta rsp failed, code:%s, tbFName:%s", tstrerror(code), target);
CTG_ERR_RET(code);
}
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);
@ -600,7 +621,7 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
if (TDMT_VND_TABLE_CFG == msgType) {
SCtgTbCfgCtx* ctx = (SCtgTbCfgCtx*)pTask->taskCtx;
pName = ctx->pName;
} else if (TDMT_VND_TABLE_META == msgType) {
} else if (TDMT_VND_TABLE_META == msgType || TDMT_VND_TABLE_NAME == msgType) {
if (CTG_TASK_GET_TB_META_BATCH == pTask->type) {
SCtgTbMetasCtx* ctx = (SCtgTbMetasCtx*)pTask->taskCtx;
SCtgFetch* fetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
@ -614,6 +635,10 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
ctgError("fail to get %d SName, totalTables:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pTbReq->pTables));
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
} else if (CTG_TASK_GET_TB_NAME == pTask->type) {
SCtgTbNamesCtx* ctx = (SCtgTbNamesCtx*)pTask->taskCtx;
SCtgFetch* fetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
CTG_ERR_JRET(ctgGetFetchName(ctx->pNames, fetch, &pName));
} else {
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
pName = ctx->pName;
@ -677,7 +702,7 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
if (TDMT_VND_TABLE_CFG == msgType) {
SCtgTbCfgCtx* ctx = (SCtgTbCfgCtx*)pTask->taskCtx;
pName = ctx->pName;
} else if (TDMT_VND_TABLE_META == msgType) {
} else if (TDMT_VND_TABLE_META == msgType || TDMT_VND_TABLE_NAME == msgType) {
if (CTG_TASK_GET_TB_META_BATCH == pTask->type) {
SCtgTbMetasCtx* ctx = (SCtgTbMetasCtx*)pTask->taskCtx;
SCtgFetch* fetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
@ -691,6 +716,10 @@ int32_t ctgAddBatch(SCatalog* pCtg, int32_t vgId, SRequestConnInfo* pConn, SCtgT
ctgError("fail to get %d SName, totalTables:%d", pFetch->tbIdx, (int32_t)taosArrayGetSize(pTbReq->pTables));
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
}
} else if (CTG_TASK_GET_TB_NAME == pTask->type) {
SCtgTbMetasCtx* ctx = (SCtgTbMetasCtx*)pTask->taskCtx;
SCtgFetch* fetch = taosArrayGet(ctx->pFetchs, tReq->msgIdx);
CTG_ERR_JRET(ctgGetFetchName(ctx->pNames, fetch, &pName));
} else {
SCtgTbMetaCtx* ctx = (SCtgTbMetaCtx*)pTask->taskCtx;
pName = ctx->pName;
@ -1337,7 +1366,7 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
SCtgTask* pTask = tReq ? tReq->pTask : NULL;
char dbFName[TSDB_DB_FNAME_LEN];
(void)tNameGetFullDbName(pTableName, dbFName);
int32_t reqType = TDMT_VND_TABLE_META;
int32_t reqType = (pTask && pTask->type == CTG_TASK_GET_TB_NAME ? TDMT_VND_TABLE_NAME : TDMT_VND_TABLE_META);
char tbFName[TSDB_TABLE_FNAME_LEN];
(void)sprintf(tbFName, "%s.%s", dbFName, pTableName->tname);
void* (*mallocFp)(int64_t) = pTask ? (MallocType)taosMemoryMalloc : (MallocType)rpcMallocCont;
@ -1346,10 +1375,12 @@ int32_t ctgGetTbMetaFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
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);
SBuildTableInput bInput = {
.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char*)tNameGetTableName(pTableName)};
char* msg = NULL;
int32_t msgLen = 0;
SBuildTableInput bInput = {.vgId = vgroupInfo->vgId,
.option = reqType == TDMT_VND_TABLE_NAME ? REQ_OPT_TBUID : REQ_OPT_TBNAME,
.dbFName = dbFName,
.tbName = (char*)tNameGetTableName(pTableName)};
char* msg = NULL;
int32_t msgLen = 0;
int32_t code = queryBuildMsg[TMSG_INDEX(reqType)](&bInput, &msg, 0, &msgLen, mallocFp);
if (code) {

41
source/libs/catalog/src/ctgUtil.c
View File

@ -111,6 +111,8 @@ char* ctgTaskTypeStr(CTG_TASK_TYPE type) {
return "[get table TSMA]";
case CTG_TASK_GET_TSMA:
return "[get TSMA]";
case CTG_TASK_GET_TB_NAME:
return "[bget table name]";
default:
return "unknown";
}
@ -584,7 +586,8 @@ void ctgFreeMsgCtx(SCtgMsgCtx* pCtx) {
break;
}
case TDMT_VND_TABLE_META:
case TDMT_MND_TABLE_META: {
case TDMT_MND_TABLE_META:
case TDMT_VND_TABLE_NAME: {
STableMetaOutput* pOut = (STableMetaOutput*)pCtx->out;
taosMemoryFree(pOut->tbMeta);
taosMemoryFreeClear(pCtx->out);
@ -844,6 +847,15 @@ void ctgFreeTaskRes(CTG_TASK_TYPE type, void** pRes) {
*pRes = NULL;
break;
}
case CTG_TASK_GET_TB_NAME: {
SArray* pArray = (SArray*)*pRes;
int32_t num = taosArrayGetSize(pArray);
for (int32_t i = 0; i < num; ++i) {
ctgFreeBatchMeta(taosArrayGet(pArray, i));
}
*pRes = NULL; // no need to free it
break;
}
default:
qError("invalid task type %d", type);
break;
@ -906,6 +918,11 @@ void ctgFreeSubTaskRes(CTG_TASK_TYPE type, void** pRes) {
*pRes = NULL;
break;
}
case CTG_TASK_GET_TB_NAME: {
taosArrayDestroyEx(*pRes, ctgFreeBatchMeta);
*pRes = NULL;
break;
}
default:
qError("invalid task type %d", type);
break;
@ -1016,6 +1033,21 @@ void ctgFreeTaskCtx(SCtgTask* pTask) {
taosMemoryFreeClear(pTask->taskCtx);
break;
}
case CTG_TASK_GET_TB_NAME: {
SCtgTbNamesCtx* taskCtx = (SCtgTbNamesCtx*)pTask->taskCtx;
taosArrayDestroyEx(taskCtx->pResList, ctgFreeBatchMeta);
taosArrayDestroy(taskCtx->pFetchs);
// NO NEED TO FREE pNames
taosArrayDestroyEx(pTask->msgCtxs, (FDelete)ctgFreeTbMetasMsgCtx);
if (pTask->msgCtx.lastOut) {
ctgFreeSTableMetaOutput((STableMetaOutput*)pTask->msgCtx.lastOut);
pTask->msgCtx.lastOut = NULL;
}
taosMemoryFreeClear(pTask->taskCtx);
break;
}
default:
qError("invalid task type %d", pTask->type);
break;
@ -1140,6 +1172,8 @@ int32_t ctgGenerateVgList(SCatalog* pCtg, SHashObj* vgHash, SArray** pList) {
pIter = taosHashIterate(vgHash, pIter);
}
taosArraySort(vgList, ctgVgInfoComp);
*pList = vgList;
ctgDebug("Got vgList from cache, vgNum:%d", vgNum);
@ -1533,7 +1567,8 @@ int32_t ctgMakeVgArray(SDBVgInfo* dbInfo) {
}
if (dbInfo->vgHash && NULL == dbInfo->vgArray) {
dbInfo->vgArray = taosArrayInit(100, sizeof(SVgroupInfo));
int32_t vgSize = taosHashGetSize(dbInfo->vgHash);
dbInfo->vgArray = taosArrayInit(vgSize, sizeof(SVgroupInfo));
if (NULL == dbInfo->vgArray) {
CTG_ERR_RET(terrno);
}
@ -1669,7 +1704,7 @@ int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes) {
}
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo* pMsgSendInfo, int32_t msgType, char* dbFName, int32_t vgId) {
if (msgType == TDMT_VND_TABLE_META || msgType == TDMT_VND_TABLE_CFG || msgType == TDMT_VND_BATCH_META) {
if (msgType == TDMT_VND_TABLE_META || msgType == TDMT_VND_TABLE_CFG || msgType == TDMT_VND_BATCH_META || msgType == TDMT_VND_TABLE_NAME) {
pMsgSendInfo->target.type = TARGET_TYPE_VNODE;
pMsgSendInfo->target.vgId = vgId;
pMsgSendInfo->target.dbFName = taosStrdup(dbFName);

47
source/libs/monitorfw/src/taos_metric_formatter_custom.c
View File

@ -88,46 +88,45 @@ int taos_metric_formatter_load_sample_new(taos_metric_formatter_t *self, taos_me
char* value = *(pair + 1);
SJson* tag = tjsonCreateObject();
if (tjsonAddStringToObject(tag, "name", key) != 0) {
if ((r = tjsonAddStringToObject(tag, "name", key)) != 0) {
taosMemoryFreeClear(arr);
taosMemoryFreeClear(keyvalue);
taosMemoryFreeClear(keyvalues);
return 1;
return r;
}
if (tjsonAddStringToObject(tag, "value", value) != 0) {
if ((r = tjsonAddStringToObject(tag, "value", value)) != 0) {
taosMemoryFreeClear(arr);
taosMemoryFreeClear(keyvalue);
taosMemoryFreeClear(keyvalues);
return 1;
return r;
}
if (tjsonAddItemToArray(arrayTag, tag) != 0) {
if ((r = tjsonAddItemToArray(arrayTag, tag)) != 0) {
taosMemoryFreeClear(arr);
taosMemoryFreeClear(keyvalue);
taosMemoryFreeClear(keyvalues);
return 1;
return r;
}
}
if (tjsonAddItemToObject(item, "tags", arrayTag) != 0) {
if ((r = tjsonAddItemToObject(item, "tags", arrayTag)) != 0) {
taosMemoryFreeClear(arr);
taosMemoryFreeClear(keyvalue);
taosMemoryFreeClear(keyvalues);
return 1;
return r;
}
metrics = tjsonCreateArray();
if (tjsonAddItemToObject(item, "metrics", metrics) != 0) {
if ((r = tjsonAddItemToObject(item, "metrics", metrics)) != 0) {
taosMemoryFreeClear(arr);
taosMemoryFreeClear(keyvalue);
taosMemoryFreeClear(keyvalues);
return 1;
return r;
}
if (tjsonAddItemToArray(arrayMetricGroups, item) != 0) {
if ((r = tjsonAddItemToArray(arrayMetricGroups, item)) != 0) {
taosMemoryFreeClear(arr);
taosMemoryFreeClear(keyvalue);
taosMemoryFreeClear(keyvalues);
return 1;
return r;
}
}
else{
@ -139,20 +138,20 @@ int taos_metric_formatter_load_sample_new(taos_metric_formatter_t *self, taos_me
taosMemoryFreeClear(keyvalues);
SJson* metric = tjsonCreateObject();
if (tjsonAddStringToObject(metric, "name", metricName) != 0) return 1;
if ((r = tjsonAddStringToObject(metric, "name", metricName)) != 0) return r;
double old_value = 0;
#define USE_EXCHANGE
#ifdef USE_EXCHANGE
if (taos_metric_sample_exchange(sample, 0, &old_value) != 0) return 1;
if ((r = taos_metric_sample_exchange(sample, 0, &old_value)) != 0) return r;
#else
old_value = sample->r_value;
taos_metric_sample_set(sample, 0);
#endif
if (tjsonAddDoubleToObject(metric, "value", old_value) != 0) return 1;
if (tjsonAddDoubleToObject(metric, "type", metric_type) != 0) return 1;
if (tjsonAddItemToArray(metrics, metric) != 0) return 1;
if ((r = tjsonAddDoubleToObject(metric, "value", old_value)) != 0) return r;
if ((r = tjsonAddDoubleToObject(metric, "type", metric_type)) != 0) return r;
if ((r = tjsonAddItemToArray(metrics, metric)) != 0) return r;
return 0;
}
@ -180,8 +179,7 @@ int taos_metric_formatter_load_metric_new(taos_metric_formatter_t *self, taos_me
SJson* table = tjsonGetArrayItem(tableArray, i);
char tableName[MONITOR_TABLENAME_LEN] = {0};
r = tjsonGetStringValue(table, "name", tableName);
if (r) {
if ((r = tjsonGetStringValue(table, "name", tableName)) != 0) {
taosMemoryFreeClear(name);
return r;
}
@ -195,15 +193,13 @@ int taos_metric_formatter_load_metric_new(taos_metric_formatter_t *self, taos_me
if(!isFound){
table = tjsonCreateObject();
r = tjsonAddStringToObject(table, "name", arr[0]);
if (r) {
if ((r = tjsonAddStringToObject(table, "name", arr[0])) != 0) {
taosMemoryFreeClear(name);
return r;
}
arrayMetricGroups = tjsonCreateArray();
r = tjsonAddItemToObject(table, "metric_groups", arrayMetricGroups);
if (r) {
if ((r = tjsonAddItemToObject(table, "metric_groups", arrayMetricGroups)) != 0) {
taosMemoryFreeClear(name);
return r;
}
@ -225,8 +221,7 @@ int taos_metric_formatter_load_metric_new(taos_metric_formatter_t *self, taos_me
}
if(!isFound && sample_count > 0){
r = tjsonAddItemToArray(tableArray, table);
if (r) {
if ((r = tjsonAddItemToArray(tableArray, table)) != 0) {
taosMemoryFreeClear(name);
return r;
}

7
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -6040,6 +6040,7 @@ static int32_t jsonToDropTableStmt(const SJson* pJson, void* pObj) {
static const char* jkDropSuperTableStmtDbName = "DbName";
static const char* jkDropSuperTableStmtTableName = "TableName";
static const char* jkDropSuperTableStmtIgnoreNotExists = "IgnoreNotExists";
static const char* jkDropSuperTableStmtwithOpt = "withOpt";
static int32_t dropStableStmtToJson(const void* pObj, SJson* pJson) {
const SDropSuperTableStmt* pNode = (const SDropSuperTableStmt*)pObj;
@ -6051,6 +6052,9 @@ static int32_t dropStableStmtToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddBoolToObject(pJson, jkDropSuperTableStmtIgnoreNotExists, pNode->ignoreNotExists);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddBoolToObject(pJson, jkDropSuperTableStmtwithOpt, pNode->withOpt);
}
return code;
}
@ -6065,6 +6069,9 @@ static int32_t jsonToDropStableStmt(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkDropSuperTableStmtIgnoreNotExists, &pNode->ignoreNotExists);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkDropSuperTableStmtwithOpt, &pNode->withOpt);
}
return code;
}

4
source/libs/parser/inc/parAst.h
View File

@ -211,8 +211,8 @@ SNode* createCreateSubTableFromFileClause(SAstCreateContext* pCxt, bool ignoreEx
SNodeList* pSpecificTags, const SToken* pFilePath);
SNode* createCreateMultiTableStmt(SAstCreateContext* pCxt, SNodeList* pSubTables);
SNode* createDropTableClause(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pRealTable);
SNode* createDropTableStmt(SAstCreateContext* pCxt, SNodeList* pTables);
SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pRealTable);
SNode* createDropTableStmt(SAstCreateContext* pCxt, bool withOpt, SNodeList* pTables);
SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool withOpt, bool ignoreNotExists, SNode* pRealTable);
SNode* createAlterTableModifyOptions(SAstCreateContext* pCxt, SNode* pRealTable, SNode* pOptions);
SNode* createAlterTableAddModifyCol(SAstCreateContext* pCxt, SNode* pRealTable, int8_t alterType, SToken* pColName,
SDataType dataType);

1
source/libs/parser/inc/parTranslater.h
View File

@ -43,6 +43,7 @@ typedef struct STranslateContext {
bool createStream;
bool stableQuery;
bool showRewrite;
bool withOpt;
SNode* pPrevRoot;
SNode* pPostRoot;
bool dual; // whether select stmt without from stmt, true for without.

6
source/libs/parser/inc/parUtil.h
View File

@ -111,7 +111,8 @@ typedef struct SParseMetaCache {
SHashObj* pTableCfg; // key is tbFName, element is STableCfg*
SHashObj* pViews; // key is viewFName, element is SViewMeta*
SHashObj* pTableTSMAs; // key is tbFName, elements are SArray<STableTSMAInfo*>
SHashObj* pTSMAs; // key is tsmaFName, elemetns are STableTSMAInfo*
SHashObj* pTSMAs; // key is tsmaFName, elements are STableTSMAInfo*
SHashObj* pTableName; // key is tbFUid, elements is STableMeta*(append with tbName)
SArray* pDnodes; // element is SEpSet
bool dnodeRequired;
} SParseMetaCache;
@ -141,6 +142,7 @@ int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalog
int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMetaData, SParseMetaCache* pMetaCache);
int32_t reserveTableMetaInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveTableMetaInCacheExt(const SName* pName, SParseMetaCache* pMetaCache);
int32_t reserveTableUidInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveViewMetaInCache(int32_t acctId, const char* pDb, const char* pView, SParseMetaCache* pMetaCache);
int32_t reserveViewMetaInCacheExt(const SName* pName, SParseMetaCache* pMetaCache);
int32_t reserveDbVgInfoInCache(int32_t acctId, const char* pDb, SParseMetaCache* pMetaCache);
@ -159,10 +161,12 @@ int32_t reserveDnodeRequiredInCache(SParseMetaCache* pMetaCache);
int32_t reserveTableTSMAInfoInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveTSMAInfoInCache(int32_t acctId, const char* pDb, const char* pTsmaName, SParseMetaCache* pMetaCache);
int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta);
int32_t getTableNameFromCache(SParseMetaCache* pMetaCache, const SName* pName, char* pTbName);
int32_t getViewMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta);
int32_t buildTableMetaFromViewMeta(STableMeta** pMeta, SViewMeta* pViewMeta);
int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo);
int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup);
int32_t getDbTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup);
int32_t getDbVgVersionFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, int32_t* pVersion, int64_t* pDbId,
int32_t* pTableNum, int64_t* pStateTs);
int32_t getDbCfgFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SDbCfgInfo* pInfo);

17
source/libs/parser/inc/sql.y
View File

@ -124,8 +124,8 @@ sysinfo_opt(A) ::= .
sysinfo_opt(A) ::= SYSINFO NK_INTEGER(B). { A = taosStr2Int8(B.z, NULL, 10); }
/************************************************ grant/revoke ********************************************************/
cmd ::= GRANT privileges(A) ON priv_level(B) with_opt(D) TO user_name(C). { pCxt->pRootNode = createGrantStmt(pCxt, A, &B, &C, D); }
cmd ::= REVOKE privileges(A) ON priv_level(B) with_opt(D) FROM user_name(C). { pCxt->pRootNode = createRevokeStmt(pCxt, A, &B, &C, D); }
cmd ::= GRANT privileges(A) ON priv_level(B) with_clause_opt(D) TO user_name(C). { pCxt->pRootNode = createGrantStmt(pCxt, A, &B, &C, D); }
cmd ::= REVOKE privileges(A) ON priv_level(B) with_clause_opt(D) FROM user_name(C). { pCxt->pRootNode = createRevokeStmt(pCxt, A, &B, &C, D); }
%type privileges { int64_t }
%destructor privileges { }
@ -151,8 +151,8 @@ priv_level(A) ::= db_name(B) NK_DOT NK_STAR(C).
priv_level(A) ::= db_name(B) NK_DOT table_name(C). { A.first = B; A.second = C; }
priv_level(A) ::= topic_name(B). { A.first = B; A.second = nil_token; }
with_opt(A) ::= . { A = NULL; }
with_opt(A) ::= WITH search_condition(B). { A = B; }
with_clause_opt(A) ::= . { A = NULL; }
with_clause_opt(A) ::= WITH search_condition(B). { A = B; }
/************************************************ create encrypt_key *********************************************/
cmd ::= CREATE ENCRYPT_KEY NK_STRING(A). { pCxt->pRootNode = createEncryptKeyStmt(pCxt, &A); }
@ -357,8 +357,8 @@ cmd ::= CREATE TABLE not_exists_opt(B) USING full_table_name(C)
NK_LP tag_list_opt(D) NK_RP FILE NK_STRING(E). { pCxt->pRootNode = createCreateSubTableFromFileClause(pCxt, B, C, D, &E); }
cmd ::= CREATE STABLE not_exists_opt(A) full_table_name(B)
NK_LP column_def_list(C) NK_RP tags_def(D) table_options(E). { pCxt->pRootNode = createCreateTableStmt(pCxt, A, B, C, D, E); }
cmd ::= DROP TABLE multi_drop_clause(A). { pCxt->pRootNode = createDropTableStmt(pCxt, A); }
cmd ::= DROP STABLE exists_opt(A) full_table_name(B). { pCxt->pRootNode = createDropSuperTableStmt(pCxt, A, B); }
cmd ::= DROP TABLE with_opt(A) multi_drop_clause(B). { pCxt->pRootNode = createDropTableStmt(pCxt, A, B); }
cmd ::= DROP STABLE with_opt(A) exists_opt(B) full_table_name(C). { pCxt->pRootNode = createDropSuperTableStmt(pCxt, A, B, C); }
cmd ::= ALTER TABLE alter_table_clause(A). { pCxt->pRootNode = A; }
cmd ::= ALTER STABLE alter_table_clause(A). { pCxt->pRootNode = setAlterSuperTableType(A); }
@ -399,6 +399,11 @@ multi_drop_clause(A) ::= multi_drop_clause(B) NK_COMMA drop_table_clause(C).
drop_table_clause(A) ::= exists_opt(B) full_table_name(C). { A = createDropTableClause(pCxt, B, C); }
%type with_opt { bool }
%destructor with_opt { }
with_opt(A) ::= . { A = false; }
with_opt(A) ::= WITH. { A = true; }
%type specific_cols_opt { SNodeList* }
%destructor specific_cols_opt { nodesDestroyList($$); }
specific_cols_opt(A) ::= . { A = NULL; }

6
source/libs/parser/src/parAstCreater.c
View File

@ -2355,19 +2355,20 @@ _err:
return NULL;
}
SNode* createDropTableStmt(SAstCreateContext* pCxt, SNodeList* pTables) {
SNode* createDropTableStmt(SAstCreateContext* pCxt, bool withOpt, SNodeList* pTables) {
CHECK_PARSER_STATUS(pCxt);
SDropTableStmt* pStmt = NULL;
pCxt->errCode = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT, (SNode**)&pStmt);
CHECK_MAKE_NODE(pStmt);
pStmt->pTables = pTables;
pStmt->withOpt = withOpt;
return (SNode*)pStmt;
_err:
nodesDestroyList(pTables);
return NULL;
}
SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool ignoreNotExists, SNode* pRealTable) {
SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool withOpt, bool ignoreNotExists, SNode* pRealTable) {
CHECK_PARSER_STATUS(pCxt);
SDropSuperTableStmt* pStmt = NULL;
pCxt->errCode = nodesMakeNode(QUERY_NODE_DROP_SUPER_TABLE_STMT, (SNode**)&pStmt);
@ -2375,6 +2376,7 @@ SNode* createDropSuperTableStmt(SAstCreateContext* pCxt, bool ignoreNotExists, S
strcpy(pStmt->dbName, ((SRealTableNode*)pRealTable)->table.dbName);
strcpy(pStmt->tableName, ((SRealTableNode*)pRealTable)->table.tableName);
pStmt->ignoreNotExists = ignoreNotExists;
pStmt->withOpt = withOpt;
nodesDestroyNode(pRealTable);
return (SNode*)pStmt;
_err:

28
source/libs/parser/src/parAstParser.c
View File

@ -379,13 +379,24 @@ static int32_t collectMetaKeyFromDropTable(SCollectMetaKeyCxt* pCxt, SDropTableS
SNode* pNode = NULL;
FOREACH(pNode, pStmt->pTables) {
SDropTableClause* pClause = (SDropTableClause*)pNode;
code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
code = reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
}
if (TSDB_CODE_SUCCESS == code) {
code = reserveUserAuthInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->pUser, pClause->dbName,
pClause->tableName, AUTH_TYPE_WRITE, pCxt->pMetaCache);
if (pStmt->withOpt) {
code = reserveTableUidInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, pClause->dbName, pCxt->pMetaCache);
}
if (TSDB_CODE_SUCCESS == code) {
code = reserveDbCfgInCache(pCxt->pParseCxt->acctId, pClause->dbName, pCxt->pMetaCache);
}
} else {
code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
code =
reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
}
if (TSDB_CODE_SUCCESS == code) {
code = reserveUserAuthInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->pUser, pClause->dbName,
pClause->tableName, AUTH_TYPE_WRITE, pCxt->pMetaCache);
}
}
if (TSDB_CODE_SUCCESS != code) {
break;
@ -395,6 +406,9 @@ static int32_t collectMetaKeyFromDropTable(SCollectMetaKeyCxt* pCxt, SDropTableS
}
static int32_t collectMetaKeyFromDropStable(SCollectMetaKeyCxt* pCxt, SDropSuperTableStmt* pStmt) {
if (pStmt->withOpt) {
return reserveTableUidInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
}
return reserveUserAuthInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->pUser, pStmt->dbName, pStmt->tableName,
AUTH_TYPE_WRITE, pCxt->pMetaCache);
}

8
source/libs/parser/src/parAuthenticator.c
View File

@ -290,7 +290,10 @@ static int32_t authCreateMultiTable(SAuthCxt* pCxt, SCreateMultiTablesStmt* pStm
static int32_t authDropTable(SAuthCxt* pCxt, SDropTableStmt* pStmt) {
int32_t code = TSDB_CODE_SUCCESS;
SNode* pNode = NULL;
if (pStmt->withOpt && !pCxt->pParseCxt->isSuperUser) {
return TSDB_CODE_PAR_PERMISSION_DENIED;
}
SNode* pNode = NULL;
FOREACH(pNode, pStmt->pTables) {
SDropTableClause* pClause = (SDropTableClause*)pNode;
code = checkAuth(pCxt, pClause->dbName, pClause->tableName, AUTH_TYPE_WRITE, NULL);
@ -302,6 +305,9 @@ static int32_t authDropTable(SAuthCxt* pCxt, SDropTableStmt* pStmt) {
}
static int32_t authDropStable(SAuthCxt* pCxt, SDropSuperTableStmt* pStmt) {
if (pStmt->withOpt && !pCxt->pParseCxt->isSuperUser) {
return TSDB_CODE_PAR_PERMISSION_DENIED;
}
return checkAuth(pCxt, pStmt->dbName, pStmt->tableName, AUTH_TYPE_WRITE, NULL);
}

186
source/libs/parser/src/parTranslater.c
View File

@ -478,6 +478,99 @@ static int32_t getViewMetaImpl(SParseContext* pParCxt, SParseMetaCache* pMetaCac
}
#endif
static int32_t getTargetNameImpl(SParseContext* pParCxt, SParseMetaCache* pMetaCache, const SName* pName,
char* pTbName) {
int32_t code = TSDB_CODE_SUCCESS;
if (pParCxt->async) {
code = getTableNameFromCache(pMetaCache, pName, pTbName);
} else {
code = TSDB_CODE_PAR_INTERNAL_ERROR;
}
if (TSDB_CODE_SUCCESS != code && TSDB_CODE_PAR_TABLE_NOT_EXIST != code) {
parserError("0x%" PRIx64 " catalogGetTableMeta error, code:%s, dbName:%s, tbName:%s", pParCxt->requestId,
tstrerror(code), pName->dbname, pName->tname);
}
return code;
}
static int32_t getTargetName(STranslateContext* pCxt, const SName* pName, char* pTbName) {
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) {
code = getTargetNameImpl(pParCxt, pCxt->pMetaCache, pName, pTbName);
}
if (TSDB_CODE_SUCCESS != code && TSDB_CODE_PAR_TABLE_NOT_EXIST != code) {
parserError("0x%" PRIx64 " catalogGetTableMeta error, code:%s, dbName:%s, tbName:%s", pCxt->pParseCxt->requestId,
tstrerror(code), pName->dbname, pName->tname);
}
return code;
}
static int32_t rewriteDropTableWithMetaCache(STranslateContext* pCxt) {
int32_t code = TSDB_CODE_SUCCESS;
SParseContext* pParCxt = pCxt->pParseCxt;
SParseMetaCache* pMetaCache = pCxt->pMetaCache;
int32_t tbMetaSize = taosHashGetSize(pMetaCache->pTableMeta);
int32_t tbMetaExSize = taosHashGetSize(pMetaCache->pTableName);
if (tbMetaSize > 0 || tbMetaExSize <= 0) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
if (!pMetaCache->pTableMeta &&
!(pMetaCache->pTableMeta =
taosHashInit(tbMetaExSize, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK))) {
return TSDB_CODE_OUT_OF_MEMORY;
}
SMetaRes** ppMetaRes = NULL;
char dbName[TSDB_DB_NAME_LEN] = {0};
while ((ppMetaRes = taosHashIterate(pMetaCache->pTableName, ppMetaRes))) {
if (!(*ppMetaRes)) {
taosHashCancelIterate(pMetaCache->pTableName, ppMetaRes);
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
char* pKey = taosHashGetKey(ppMetaRes, NULL);
STableMeta* pMeta = (STableMeta*)(*ppMetaRes)->pRes;
if (!pMeta) {
taosHashCancelIterate(pMetaCache->pTableName, ppMetaRes);
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
char* pDbStart = strstr(pKey, ".");
char* pDbEnd = pDbStart ? strstr(pDbStart + 1, ".") : NULL;
if (!pDbEnd) {
taosHashCancelIterate(pMetaCache->pTableName, ppMetaRes);
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
tstrncpy(dbName, pDbStart + 1, pDbEnd - pDbStart);
int32_t metaSize =
sizeof(STableMeta) + sizeof(SSchema) * (pMeta->tableInfo.numOfColumns + pMeta->tableInfo.numOfTags);
int32_t schemaExtSize =
(useCompress(pMeta->tableType) && pMeta->schemaExt) ? sizeof(SSchemaExt) * pMeta->tableInfo.numOfColumns : 0;
const char* pTbName = (const char*)pMeta + metaSize + schemaExtSize;
SName name = {0};
toName(pParCxt->acctId, dbName, pTbName, &name);
char fullName[TSDB_TABLE_FNAME_LEN];
code = tNameExtractFullName(&name, fullName);
if (TSDB_CODE_SUCCESS != code) {
taosHashCancelIterate(pMetaCache->pTableName, ppMetaRes);
return code;
}
if ((code = taosHashPut(pMetaCache->pTableMeta, fullName, strlen(fullName), ppMetaRes, POINTER_BYTES))) {
taosHashCancelIterate(pMetaCache->pTableName, ppMetaRes);
return code;
}
}
return code;
}
int32_t getTargetMetaImpl(SParseContext* pParCxt, SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta,
bool couldBeView) {
int32_t code = TSDB_CODE_SUCCESS;
@ -617,7 +710,11 @@ static int32_t getTableHashVgroupImpl(STranslateContext* pCxt, const SName* pNam
}
if (TSDB_CODE_SUCCESS == code) {
if (pParCxt->async) {
code = getTableVgroupFromCache(pCxt->pMetaCache, pName, pInfo);
if(pCxt->withOpt) {
code = getDbTableVgroupFromCache(pCxt->pMetaCache, pName, pInfo);
} else {
code = getTableVgroupFromCache(pCxt->pMetaCache, pName, pInfo);
}
} else {
SRequestConnInfo conn = {.pTrans = pParCxt->pTransporter,
.requestId = pParCxt->requestId,
@ -14460,12 +14557,53 @@ int32_t serializeVgroupsDropTableBatch(SHashObj* pVgroupHashmap, SArray** pOut)
return code;
}
static int32_t rewriteDropTableWithOpt(STranslateContext* pCxt, SQuery* pQuery) {
int32_t code = TSDB_CODE_SUCCESS;
SDropTableStmt* pStmt = (SDropTableStmt*)pQuery->pRoot;
if (!pStmt->withOpt) return code;
pCxt->withOpt = true;
SNode* pNode = NULL;
char pTableName[TSDB_TABLE_NAME_LEN] = {0};
FOREACH(pNode, pStmt->pTables) {
SDropTableClause* pClause = (SDropTableClause*)pNode;
if (IS_SYS_DBNAME(pClause->dbName)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_TSC_INVALID_OPERATION,
"Cannot drop table of system database: `%s`.`%s`", pClause->dbName,
pClause->tableName);
}
for (int32_t i = 0; i < TSDB_TABLE_NAME_LEN; i++) {
if (pClause->tableName[i] == '\0') {
break;
}
if (!isdigit(pClause->tableName[i])) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_TABLE_NOT_EXIST, "Table does not exist: `%s`.`%s`",
pClause->dbName, pClause->tableName);
}
}
SName name = {0};
toName(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, &name);
int32_t code = getTargetName(pCxt, &name, pTableName);
if (TSDB_CODE_SUCCESS != code) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, code, "%s: db:`%s`, tbuid:`%s`", tstrerror(code), pClause->dbName,
pClause->tableName);
}
tstrncpy(pClause->tableName, pTableName, TSDB_TABLE_NAME_LEN); // rewrite table uid to table name
}
code = rewriteDropTableWithMetaCache(pCxt);
TAOS_RETURN(code);
}
static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
SDropTableStmt* pStmt = (SDropTableStmt*)pQuery->pRoot;
int8_t tableType;
SNode* pNode;
SArray* pTsmas = NULL;
TAOS_CHECK_RETURN(rewriteDropTableWithOpt(pCxt, pQuery));
SHashObj* pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (NULL == pVgroupHashmap) {
return terrno;
@ -14482,8 +14620,10 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
return code;
}
if (tableType == TSDB_SUPER_TABLE && LIST_LENGTH(pStmt->pTables) > 1) {
taosHashCleanup(pVgroupHashmap);
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_DROP_STABLE);
}
if (pCxt->withOpt) continue;
if (pCxt->pMetaCache) code = getTableTsmasFromCache(pCxt->pMetaCache, &name, &pTsmas);
if (TSDB_CODE_SUCCESS != code) {
taosHashCleanup(pVgroupHashmap);
@ -14545,6 +14685,47 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
return rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
}
static int32_t rewriteDropSuperTablewithOpt(STranslateContext* pCxt, SQuery* pQuery) {
int32_t code = TSDB_CODE_SUCCESS;
SDropSuperTableStmt* pStmt = (SDropSuperTableStmt*)pQuery->pRoot;
if (!pStmt->withOpt) return code;
pCxt->withOpt = true;
if (IS_SYS_DBNAME(pStmt->dbName)) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_TSC_INVALID_OPERATION,
"Cannot drop table of system database: `%s`.`%s`", pStmt->dbName, pStmt->tableName);
}
for (int32_t i = 0; i < TSDB_TABLE_NAME_LEN; i++) {
if (pStmt->tableName[i] == '\0') {
break;
}
if (!isdigit(pStmt->tableName[i])) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_TABLE_NOT_EXIST, "Table does not exist: `%s`.`%s`",
pStmt->dbName, pStmt->tableName);
}
}
char pTableName[TSDB_TABLE_NAME_LEN] = {0};
SName name = {0};
toName(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, &name);
code = getTargetName(pCxt, &name, pTableName);
if (TSDB_CODE_SUCCESS != code) {
return generateSyntaxErrMsgExt(&pCxt->msgBuf, code, "%s: db:`%s`, tbuid:`%s`", tstrerror(code), pStmt->dbName,
pStmt->tableName);
}
tstrncpy(pStmt->tableName, pTableName, TSDB_TABLE_NAME_LEN); // rewrite table uid to table name
code = rewriteDropTableWithMetaCache(pCxt);
TAOS_RETURN(code);
}
static int32_t rewriteDropSuperTable(STranslateContext* pCxt, SQuery* pQuery) {
TAOS_CHECK_RETURN(rewriteDropSuperTablewithOpt(pCxt, pQuery));
TAOS_RETURN(0);
}
static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SName tbName = {0};
@ -15616,6 +15797,9 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
case QUERY_NODE_DROP_TABLE_STMT:
code = rewriteDropTable(pCxt, pQuery);
break;
case QUERY_NODE_DROP_SUPER_TABLE_STMT:
code = rewriteDropSuperTable(pCxt, pQuery);
break;
case QUERY_NODE_ALTER_TABLE_STMT:
code = rewriteAlterTable(pCxt, pQuery);
break;

67
source/libs/parser/src/parUtil.c
View File

@ -804,6 +804,9 @@ int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalog
if (TSDB_CODE_SUCCESS == code) {
code = buildTableReqFromDb(pMetaCache->pTSMAs, &pCatalogReq->pTSMAs);
}
if (TSDB_CODE_SUCCESS == code) {
code = buildTableReqFromDb(pMetaCache->pTableName, &pCatalogReq->pTableName);
}
#ifdef TD_ENTERPRISE
if (TSDB_CODE_SUCCESS == code) {
code = buildTableReqFromDb(pMetaCache->pTableMeta, &pCatalogReq->pView);
@ -960,6 +963,9 @@ int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMet
if (TSDB_CODE_SUCCESS == code) {
code = putDbTableDataToCache(pCatalogReq->pTSMAs, pMetaData->pTsmas, &pMetaCache->pTSMAs);
}
if (TSDB_CODE_SUCCESS == code) {
code = putDbTableDataToCache(pCatalogReq->pTableName, pMetaData->pTableMeta, &pMetaCache->pTableName);
}
#ifdef TD_ENTERPRISE
if (TSDB_CODE_SUCCESS == code) {
code = putDbTableDataToCache(pCatalogReq->pView, pMetaData->pView, &pMetaCache->pViews);
@ -1019,8 +1025,12 @@ int32_t reserveTableMetaInCacheExt(const SName* pName, SParseMetaCache* pMetaCac
return reserveTableReqInDbCache(pName->acctId, pName->dbname, pName->tname, &pMetaCache->pTableMeta);
}
int32_t reserveTableUidInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
return reserveTableReqInDbCache(acctId, pDb, pTable, &pMetaCache->pTableName);
}
int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta) {
char fullName[TSDB_TABLE_FNAME_LEN];
char fullName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(pName, fullName);
if (TSDB_CODE_SUCCESS != code) {
return code;
@ -1036,6 +1046,27 @@ int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, S
return code;
}
int32_t getTableNameFromCache(SParseMetaCache* pMetaCache, const SName* pName, char* pTbName) {
char fullName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(pName, fullName);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
const STableMeta* pMeta = NULL;
code = getMetaDataFromHash(fullName, strlen(fullName), pMetaCache->pTableName, (void**)&pMeta);
if (TSDB_CODE_SUCCESS == code) {
if (!pMeta) code = TSDB_CODE_PAR_INTERNAL_ERROR;
int32_t metaSize =
sizeof(STableMeta) + sizeof(SSchema) * (pMeta->tableInfo.numOfColumns + pMeta->tableInfo.numOfTags);
int32_t schemaExtSize =
(useCompress(pMeta->tableType) && pMeta->schemaExt) ? sizeof(SSchemaExt) * pMeta->tableInfo.numOfColumns : 0;
const char* pTableName = (const char*)pMeta + metaSize + schemaExtSize;
tstrncpy(pTbName, pTableName, TSDB_TABLE_NAME_LEN);
}
return code;
}
int32_t buildTableMetaFromViewMeta(STableMeta** pMeta, SViewMeta* pViewMeta) {
*pMeta = taosMemoryCalloc(1, sizeof(STableMeta) + pViewMeta->numOfCols * sizeof(SSchema));
if (NULL == *pMeta) {
@ -1121,6 +1152,38 @@ int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName,
return code;
}
int32_t getDbTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup) {
char fullName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(pName, fullName);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
const char* pDb = strstr(fullName, ".");
if (pDb == NULL) return TSDB_CODE_PAR_INTERNAL_ERROR;
pDb = strstr(pDb + 1, ".");
if (pDb == NULL) return TSDB_CODE_PAR_INTERNAL_ERROR;
int32_t fullDbLen = pDb - fullName;
int32_t fullTbLen = strlen(fullName);
SArray* pVgArray = NULL;
SDbCfgInfo* pDbCfg = NULL;
code = getMetaDataFromHash(fullName, fullDbLen, pMetaCache->pDbVgroup, (void**)&pVgArray);
if (TSDB_CODE_SUCCESS == code) {
code = getMetaDataFromHash(fullName, fullDbLen, pMetaCache->pDbCfg, (void**)&pDbCfg);
}
if (TSDB_CODE_SUCCESS == code) {
uint32_t hashValue =
taosGetTbHashVal(fullName, fullTbLen, pDbCfg->hashMethod, pDbCfg->hashPrefix, pDbCfg->hashSuffix);
SVgroupInfo* pVg = taosArraySearch(pVgArray, &hashValue, ctgHashValueComp, TD_EQ);
if (pVg) {
memcpy(pVgroup, pVg, sizeof(SVgroupInfo));
} else {
code = TSDB_CODE_PAR_INTERNAL_ERROR;
}
}
return code;
}
int32_t reserveDbVgVersionInCache(int32_t acctId, const char* pDb, SParseMetaCache* pMetaCache) {
return reserveDbReqInCache(acctId, pDb, &pMetaCache->pDbInfo);
}
@ -1396,11 +1459,13 @@ void destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request) {
destoryParseTablesMetaReqHash(pMetaCache->pTableVgroup);
destoryParseTablesMetaReqHash(pMetaCache->pViews);
destoryParseTablesMetaReqHash(pMetaCache->pTSMAs);
destoryParseTablesMetaReqHash(pMetaCache->pTableName);
} else {
taosHashCleanup(pMetaCache->pTableMeta);
taosHashCleanup(pMetaCache->pTableVgroup);
taosHashCleanup(pMetaCache->pViews);
taosHashCleanup(pMetaCache->pTSMAs);
taosHashCleanup(pMetaCache->pTableName);
}
taosHashCleanup(pMetaCache->pDbVgroup);
taosHashCleanup(pMetaCache->pDbCfg);

2
source/libs/parser/src/parser.c
View File

@ -351,6 +351,7 @@ void destoryCatalogReq(SCatalogReq* pCatalogReq) {
#endif
taosArrayDestroy(pCatalogReq->pTableTSMAs);
taosArrayDestroy(pCatalogReq->pTSMAs);
taosArrayDestroy(pCatalogReq->pTableName);
} else {
taosArrayDestroyEx(pCatalogReq->pTableMeta, destoryTablesReq);
taosArrayDestroyEx(pCatalogReq->pTableHash, destoryTablesReq);
@ -359,6 +360,7 @@ void destoryCatalogReq(SCatalogReq* pCatalogReq) {
#endif
taosArrayDestroyEx(pCatalogReq->pTableTSMAs, destoryTablesReq);
taosArrayDestroyEx(pCatalogReq->pTSMAs, destoryTablesReq);
taosArrayDestroyEx(pCatalogReq->pTableName, destoryTablesReq);
}
taosArrayDestroy(pCatalogReq->pUdf);
taosArrayDestroy(pCatalogReq->pIndex);

13826
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff

117
source/libs/qcom/src/querymsg.c
View File

@ -77,6 +77,7 @@ int32_t queryBuildTableMetaReqMsg(void *input, char **msg, int32_t msgSize, int3
}
STableInfoReq infoReq = {0};
infoReq.option = pInput->option;
infoReq.header.vgId = pInput->vgId;
if (pInput->dbFName) {
tstrncpy(infoReq.dbFName, pInput->dbFName, TSDB_DB_FNAME_LEN);
@ -603,6 +604,64 @@ int32_t queryCreateTableMetaFromMsg(STableMetaRsp *msg, bool isStb, STableMeta *
return TSDB_CODE_SUCCESS;
}
int32_t queryCreateTableMetaExFromMsg(STableMetaRsp *msg, bool isStb, STableMeta **pMeta) {
int32_t total = msg->numOfColumns + msg->numOfTags;
int32_t metaSize = sizeof(STableMeta) + sizeof(SSchema) * total;
int32_t schemaExtSize = (useCompress(msg->tableType) && msg->pSchemaExt) ? sizeof(SSchemaExt) * msg->numOfColumns : 0;
int32_t tbNameSize = strlen(msg->tbName) + 1;
STableMeta *pTableMeta = taosMemoryCalloc(1, metaSize + schemaExtSize + tbNameSize);
if (NULL == pTableMeta) {
qError("calloc size[%d] failed", metaSize);
return terrno;
}
SSchemaExt *pSchemaExt = (SSchemaExt *)((char *)pTableMeta + metaSize);
pTableMeta->vgId = isStb ? 0 : msg->vgId;
pTableMeta->tableType = isStb ? TSDB_SUPER_TABLE : msg->tableType;
pTableMeta->uid = isStb ? msg->suid : msg->tuid;
pTableMeta->suid = msg->suid;
pTableMeta->sversion = msg->sversion;
pTableMeta->tversion = msg->tversion;
pTableMeta->tableInfo.numOfTags = msg->numOfTags;
pTableMeta->tableInfo.precision = msg->precision;
pTableMeta->tableInfo.numOfColumns = msg->numOfColumns;
TAOS_MEMCPY(pTableMeta->schema, msg->pSchemas, sizeof(SSchema) * total);
if (useCompress(msg->tableType) && msg->pSchemaExt) {
pTableMeta->schemaExt = pSchemaExt;
TAOS_MEMCPY(pSchemaExt, msg->pSchemaExt, schemaExtSize);
} else {
pTableMeta->schemaExt = NULL;
}
bool hasPK = (msg->numOfColumns > 1) && (pTableMeta->schema[1].flags & COL_IS_KEY);
for (int32_t i = 0; i < msg->numOfColumns; ++i) {
pTableMeta->tableInfo.rowSize += pTableMeta->schema[i].bytes;
if (hasPK && (i > 0)) {
if ((pTableMeta->schema[i].flags & COL_IS_KEY)) {
++pTableMeta->tableInfo.numOfPKs;
} else {
hasPK = false;
}
}
}
char *pTbName = (char *)pTableMeta + metaSize + schemaExtSize;
tstrncpy(pTbName, msg->tbName, tbNameSize);
qDebug("table %s uid %" PRIx64 " meta returned, type %d vgId:%d db %s stb %s suid %" PRIx64
" sver %d tver %d"
" tagNum %d colNum %d precision %d rowSize %d",
msg->tbName, pTableMeta->uid, pTableMeta->tableType, pTableMeta->vgId, msg->dbFName, msg->stbName,
pTableMeta->suid, pTableMeta->sversion, pTableMeta->tversion, pTableMeta->tableInfo.numOfTags,
pTableMeta->tableInfo.numOfColumns, pTableMeta->tableInfo.precision, pTableMeta->tableInfo.rowSize);
*pMeta = pTableMeta;
return TSDB_CODE_SUCCESS;
}
int32_t queryProcessTableMetaRsp(void *output, char *msg, int32_t msgSize) {
int32_t code = 0;
STableMetaRsp metaRsp = {0};
@ -659,6 +718,62 @@ PROCESS_META_OVER:
return code;
}
static int32_t queryProcessTableNameRsp(void *output, char *msg, int32_t msgSize) {
int32_t code = 0;
STableMetaRsp metaRsp = {0};
if (NULL == output || NULL == msg || msgSize <= 0) {
code = TSDB_CODE_TSC_INVALID_INPUT;
goto PROCESS_NAME_OVER;
}
if (tDeserializeSTableMetaRsp(msg, msgSize, &metaRsp) != 0) {
code = TSDB_CODE_INVALID_MSG;
goto PROCESS_NAME_OVER;
}
code = queryConvertTableMetaMsg(&metaRsp);
if (code != TSDB_CODE_SUCCESS) {
goto PROCESS_NAME_OVER;
}
if (!IS_SYS_DBNAME(metaRsp.dbFName) &&
!tIsValidSchema(metaRsp.pSchemas, metaRsp.numOfColumns, metaRsp.numOfTags)) {
code = TSDB_CODE_TSC_INVALID_VALUE;
goto PROCESS_NAME_OVER;
}
STableMetaOutput *pOut = output;
strcpy(pOut->dbFName, metaRsp.dbFName);
pOut->dbId = metaRsp.dbId;
if (metaRsp.tableType == TSDB_CHILD_TABLE) {
SET_META_TYPE_BOTH_TABLE(pOut->metaType);
strcpy(pOut->ctbName, metaRsp.tbName);
strcpy(pOut->tbName, metaRsp.stbName);
pOut->ctbMeta.vgId = metaRsp.vgId;
pOut->ctbMeta.tableType = metaRsp.tableType;
pOut->ctbMeta.uid = metaRsp.tuid;
pOut->ctbMeta.suid = metaRsp.suid;
code = queryCreateTableMetaExFromMsg(&metaRsp, true, &pOut->tbMeta);
} else {
SET_META_TYPE_TABLE(pOut->metaType);
strcpy(pOut->tbName, metaRsp.tbName);
code = queryCreateTableMetaExFromMsg(&metaRsp, (metaRsp.tableType == TSDB_SUPER_TABLE), &pOut->tbMeta);
}
PROCESS_NAME_OVER:
if (code != 0) {
qError("failed to process table name rsp since %s", tstrerror(code));
}
tFreeSTableMetaRsp(&metaRsp);
return code;
}
int32_t queryProcessQnodeListRsp(void *output, char *msg, int32_t msgSize) {
SQnodeListRsp out = {0};
int32_t code = 0;
@ -880,6 +995,7 @@ int32_t queryProcessStreamProgressRsp(void* output, char* msg, int32_t msgSize)
void initQueryModuleMsgHandle() {
queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryBuildTableMetaReqMsg;
queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_NAME)] = queryBuildTableMetaReqMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryBuildTableMetaReqMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_USE_DB)] = queryBuildUseDbMsg;
queryBuildMsg[TMSG_INDEX(TDMT_MND_QNODE_LIST)] = queryBuildQnodeListMsg;
@ -898,6 +1014,7 @@ void initQueryModuleMsgHandle() {
queryBuildMsg[TMSG_INDEX(TDMT_VND_GET_STREAM_PROGRESS)] = queryBuildGetStreamProgressMsg;
queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)] = queryProcessTableMetaRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_NAME)] = queryProcessTableNameRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)] = queryProcessTableMetaRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_USE_DB)] = queryProcessUseDBRsp;
queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_QNODE_LIST)] = queryProcessQnodeListRsp;

2
tests/system-test/0-others/information_schema.py
View File

@ -222,7 +222,7 @@ class TDTestCase:
tdSql.query("select * from information_schema.ins_columns where db_name ='information_schema'")
tdLog.info(len(tdSql.queryResult))
tdSql.checkEqual(True, len(tdSql.queryResult) in range(271, 272))
tdSql.checkEqual(True, len(tdSql.queryResult) in range(272, 273))
tdSql.query("select * from information_schema.ins_columns where db_name ='performance_schema'")
tdSql.checkEqual(56, len(tdSql.queryResult))

123
tests/system-test/1-insert/drop.py
View File

@ -49,6 +49,11 @@ class TDTestCase:
'col12': 'binary(20)',
'col13': 'nchar(20)'
}
self.db_names = [ f'dbtest_0', f'dbtest_1']
self.stb_names = [ f'aa\u00bf\u200bstb0']
self.ctb_names = [ f'ctb0', 'ctb1', f'aa\u00bf\u200bctb0', f'aa\u00bf\u200bctb1']
self.ntb_names = [ f'ntb0', f'aa\u00bf\u200bntb0', f'ntb1', f'aa\u00bf\u200bntb1']
self.vgroups_opt = f'vgroups 4'
def insert_data(self,column_dict,tbname,row_num):
insert_sql = self.setsql.set_insertsql(column_dict,tbname,self.binary_str,self.nchar_str)
for i in range(row_num):
@ -116,6 +121,121 @@ class TDTestCase:
tdSql.query(f'select * from {stbname} where {k} = {self.ts}')
tdSql.checkRows(self.tbnum)
tdSql.execute(f'drop database {self.dbname}')
def drop_table_check_init(self):
for db_name in self.db_names:
tdSql.execute(f'create database if not exists {db_name} {self.vgroups_opt}')
tdSql.execute(f'use {db_name}')
for stb_name in self.stb_names:
tdSql.execute(f'create table `{stb_name}` (ts timestamp,c0 int) tags(t0 int)')
for ctb_name in self.ctb_names:
tdSql.execute(f'create table `{ctb_name}` using `{stb_name}` tags(0)')
tdSql.execute(f'insert into `{ctb_name}` values (now,1)')
for ntb_name in self.ntb_names:
tdSql.execute(f'create table `{ntb_name}` (ts timestamp,c0 int)')
tdSql.execute(f'insert into `{ntb_name}` values (now,1)')
def drop_table_check_end(self):
for db_name in self.db_names:
tdSql.execute(f'drop database {db_name}')
def drop_stable_with_check(self):
self.drop_table_check_init()
tdSql.query(f'select * from information_schema.ins_stables where db_name like "dbtest_%"')
result = tdSql.queryResult
print(result)
tdSql.checkEqual(len(result),2)
i = 0
for stb_result in result:
if i == 0:
dropTable = f'drop table with `{stb_result[1]}`.`{stb_result[10]}`,'
dropStable = f'drop stable with `{stb_result[1]}`.`{stb_result[10]}`,'
else:
dropTable += f'`{stb_result[1]}`.`{stb_result[10]}`,'
dropStable += f'`{stb_result[1]}`.`{stb_result[10]}`,'
tdLog.info(dropTable[:-1])
tdLog.info(dropStable[:-1])
tdSql.error(dropTable[:-1])
tdSql.error(dropStable[:-1])
i += 1
i = 0
for stb_result in result:
if i == 0:
tdSql.execute(f'drop table with `{stb_result[1]}`.`{stb_result[10]}`')
else:
tdSql.execute(f'drop stable with `{stb_result[1]}`.`{stb_result[10]}`')
i += 1
for i in range(30):
tdSql.query(f'select * from information_schema.ins_stables where db_name like "dbtest_%"')
if(len(tdSql.queryResult) == 0):
break
tdLog.info(f'ins_stables not empty, sleep 1s')
time.sleep(1)
tdSql.query(f'select * from information_schema.ins_stables where db_name like "dbtest_%"')
tdSql.checkRows(0)
tdSql.query(f'select * from information_schema.ins_tables where db_name like "dbtest_%"')
tdSql.checkRows(8)
tdSql.error(f'drop stable with information_schema.`ins_tables`;')
tdSql.error(f'drop stable with performance_schema.`perf_connections`;')
self.drop_table_check_end()
def drop_table_with_check(self):
self.drop_table_check_init()
tdSql.query(f'select * from information_schema.ins_tables where db_name like "dbtest_%"')
result = tdSql.queryResult
print(result)
tdSql.checkEqual(len(result),16)
dropTable = f'drop table with '
for tb_result in result:
dropTable += f'`{tb_result[1]}`.`{tb_result[5]}`,'
tdLog.info(dropTable[:-1])
tdSql.execute(dropTable[:-1])
for i in range(30):
tdSql.query(f'select * from information_schema.ins_tables where db_name like "dbtest_%"')
if(len(tdSql.queryResult) == 0):
break
tdLog.info(f'ins_tables not empty, sleep 1s')
time.sleep(1)
tdSql.query(f'select * from information_schema.ins_tables where db_name like "dbtest_%"')
tdSql.checkRows(0)
tdSql.query(f'select * from information_schema.ins_stables where db_name like "dbtest_%"')
tdSql.checkRows(2)
tdSql.error(f'drop table with information_schema.`ins_tables`;')
tdSql.error(f'drop table with performance_schema.`perf_connections`;')
self.drop_table_check_end()
def drop_table_with_check_tsma(self):
tdSql.execute(f'create database if not exists {self.dbname} {self.vgroups_opt}')
tdSql.execute(f'use {self.dbname}')
tdSql.execute(f'create table {self.dbname}.stb (ts timestamp,c0 int) tags(t0 int)')
tdSql.execute(f'create tsma stb_tsma on {self.dbname}.stb function(avg(c0),count(c0)) interval(1d)')
tdSql.execute(f'create table {self.dbname}.ctb using {self.dbname}.stb tags(0)')
tdSql.execute(f'insert into {self.dbname}.ctb values (now,1)')
tdSql.execute(f'create table {self.dbname}.ntb (ts timestamp,c0 int)')
tdSql.execute(f'create tsma ntb_tsma on {self.dbname}.ntb function(avg(c0),count(c0)) interval(1d)')
tdSql.execute(f'insert into {self.dbname}.ntb values (now,1)')
tdSql.query(f'select * from information_schema.ins_tsmas where db_name = "{self.dbname}"')
tdSql.checkRows(2)
tdSql.query(f'select * from information_schema.ins_tables where db_name = "{self.dbname}" and type="CHILD_TABLE"')
tdSql.checkRows(1)
tdSql.execute(f'drop table with {tdSql.queryResult[0][1]}.`{tdSql.queryResult[0][5]}`')
tdSql.query(f'select * from information_schema.ins_tables where db_name = "{self.dbname}" and type="CHILD_TABLE"')
tdSql.checkRows(0)
tdSql.query(f'select * from information_schema.ins_stables where db_name = "{self.dbname}"')
tdSql.checkRows(1)
tdSql.error(f'drop table with {tdSql.queryResult[0][1]}.`{tdSql.queryResult[0][10]}`')
tdSql.query(f'select * from information_schema.ins_stables where db_name = "{self.dbname}"')
tdSql.error(f'drop stable with {tdSql.queryResult[0][1]}.`{tdSql.queryResult[0][10]}`')
tdSql.query(f'select * from information_schema.ins_stables where db_name = "{self.dbname}"')
tdSql.checkRows(1)
tdSql.query(f'select * from information_schema.ins_tables where db_name = "{self.dbname}" and type="NORMAL_TABLE"')
tdSql.checkRows(1)
tdSql.execute(f'drop table with {tdSql.queryResult[0][1]}.`{tdSql.queryResult[0][5]}`')
tdSql.query(f'select * from information_schema.ins_tables where db_name = "{self.dbname}" and type="NORMAL_TABLE"')
tdSql.checkRows(0)
tdSql.query(f'select * from information_schema.ins_tsmas where db_name = "{self.dbname}"')
tsmas = tdSql.queryResult
tdSql.checkEqual(len(tsmas),2)
for tsma in tsmas:
tdSql.execute(f'drop tsma {tsma[1]}.{tsma[0]}')
tdSql.query(f'show tsmas')
tdSql.checkRows(0)
tdSql.execute(f'drop database {self.dbname}')
def drop_topic_check(self):
tdSql.execute(f'create database {self.dbname} replica {self.replicaVar} wal_retention_period 3600')
tdSql.execute(f'use {self.dbname}')
@ -161,6 +281,9 @@ class TDTestCase:
def run(self):
self.drop_ntb_check()
self.drop_stb_ctb_check()
self.drop_stable_with_check()
self.drop_table_with_check()
self.drop_table_with_check_tsma()
self.drop_topic_check()
if platform.system().lower() == 'windows':
self.drop_stream_check()