Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 e515a4ab 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/data_format

上级 af8ac6e1 d8743b72
隐藏空白更改
内联 并排
Showing with 478 addition and 518 deletion
include/libs/nodes/querynodes.h
浏览文件 @ e515a4ab
......@@ -325,6 +325,7 @@ typedef struct SQuery {
bool showRewrite;
int32_t placeholderNum;
SArray* pPlaceholderValues;
SNode* pContainPlaceholderRoot;
} SQuery;
void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);
......
include/libs/parser/parser.h
浏览文件 @ e515a4ab
......@@ -62,7 +62,7 @@ int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t
void qDestroyStmtDataBlock(void* pBlock);
STableMeta* qGetTableMetaInDataBlock(void* pDataBlock);
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId);
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx);
int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery);
int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
......@@ -77,8 +77,8 @@ 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);
int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
char* tableName, char* msgBuf, int16_t msgBufLen);
int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
#ifdef __cplusplus
......
include/libs/planner/planner.h
浏览文件 @ e515a4ab
......@@ -34,7 +34,6 @@ typedef struct SPlanContext {
bool showRewrite;
int8_t triggerType;
int64_t watermark;
int32_t placeholderNum;
char* pMsg;
int32_t msgLen;
} SPlanContext;
......@@ -48,9 +47,6 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
// @pSource one execution location of this group of datasource subplans
int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId,
bool* pEmptyResult);
// Convert to subplan to string for the scheduler to send to the executor
int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);
int32_t qStringToSubplan(const char* pStr, SSubplan** pSubplan);
......
include/util/taoserror.h
浏览文件 @ e515a4ab
......@@ -644,6 +644,8 @@ int32_t* taosGetErrno();
#define TSDB_CODE_PAR_INVALID_TIMELINE_FUNC TAOS_DEF_ERROR_CODE(0, 0x2647)
#define TSDB_CODE_PAR_INVALID_PASSWD TAOS_DEF_ERROR_CODE(0, 0x2648)
#define TSDB_CODE_PAR_INVALID_ALTER_TABLE TAOS_DEF_ERROR_CODE(0, 0x2649)
#define TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY TAOS_DEF_ERROR_CODE(0, 0x264A)
#define TSDB_CODE_PAR_INVALID_MODIFY_COL TAOS_DEF_ERROR_CODE(0, 0x264B)
//planner
#define TSDB_CODE_PLAN_INTERNAL_ERROR TAOS_DEF_ERROR_CODE(0, 0x2700)
......
source/client/src/clientImpl.c
浏览文件 @ e515a4ab
......@@ -233,8 +233,7 @@ int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArra
.pAstRoot = pQuery->pRoot,
.showRewrite = pQuery->showRewrite,
.pMsg = pRequest->msgBuf,
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
.placeholderNum = pQuery->placeholderNum};
.msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
SEpSet mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
SCatalog* pCatalog = NULL;
int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
......@@ -518,7 +517,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t
if (pRequest->code != TSDB_CODE_SUCCESS) {
const char* errorMsg =
(pRequest->code == TSDB_CODE_RPC_FQDN_ERROR) ? taos_errstr(pRequest) : tstrerror(pRequest->code);
fprintf(stderr,"failed to connect to server, reason: %s\n\n", errorMsg);
fprintf(stderr, "failed to connect to server, reason: %s\n\n", errorMsg);
terrno = pRequest->code;
destroyRequest(pRequest);
......@@ -949,7 +948,8 @@ int32_t setQueryResultFromRsp(SReqResultInfo* pResultInfo, const SRetrieveTableR
// TODO handle the compressed case
pResultInfo->totalRows += pResultInfo->numOfRows;
return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows, convertUcs4);
return setResultDataPtr(pResultInfo, pResultInfo->fields, pResultInfo->numOfCols, pResultInfo->numOfRows,
convertUcs4);
}
TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* details, int maxlen) {
......
source/client/src/clientStmt.c
浏览文件 @ e515a4ab
......@@ -67,7 +67,7 @@ int32_t stmtGetTbName(TAOS_STMT* stmt, char** tbName) {
STscStmt* pStmt = (STscStmt*)stmt;
pStmt->sql.type = STMT_TYPE_MULTI_INSERT;
if ('\0' == pStmt->bInfo.tbName[0]) {
tscError("no table name set");
STMT_ERR_RET(TSDB_CODE_TSC_STMT_TBNAME_ERROR);
......@@ -126,7 +126,7 @@ int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags,
strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1);
pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
pStmt->bInfo.tbUid = pTableMeta->uid;
pStmt->bInfo.tbSuid = pTableMeta->suid;
pStmt->bInfo.tbType = pTableMeta->tableType;
......@@ -146,18 +146,18 @@ int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockH
return TSDB_CODE_SUCCESS;
}
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, bool autoCreateTbl, SHashObj* pVgHash, SHashObj* pBlockHash) {
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, bool autoCreateTbl,
SHashObj* pVgHash, SHashObj* pBlockHash) {
STscStmt* pStmt = (STscStmt*)stmt;
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName));
STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl));
pStmt->sql.autoCreateTbl = autoCreateTbl;
return TSDB_CODE_SUCCESS;
}
int32_t stmtGetExecInfo(TAOS_STMT* stmt, SHashObj** pVgHash, SHashObj** pBlockHash) {
STscStmt* pStmt = (STscStmt*)stmt;
......@@ -172,7 +172,7 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS;
}
uint64_t uid = pStmt->bInfo.tbUid;
uint64_t uid = pStmt->bInfo.tbUid;
uint64_t cacheUid = (TSDB_CHILD_TABLE == pStmt->bInfo.tbType) ? pStmt->bInfo.tbSuid : uid;
if (taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid))) {
......@@ -180,8 +180,8 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
}
STableDataBlocks** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataBlocks* pDst = NULL;
STableDataBlocks* pDst = NULL;
STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc));
SStmtTableCache cache = {
......@@ -198,16 +198,16 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
} else {
pStmt->bInfo.boundTags = NULL;
}
return TSDB_CODE_SUCCESS;
}
int32_t stmtParseSql(STscStmt* pStmt) {
SStmtCallback stmtCb = {
.pStmt = pStmt,
.getTbNameFn = stmtGetTbName,
.setInfoFn = stmtUpdateInfo,
.getExecInfoFn = stmtGetExecInfo,
.pStmt = pStmt,
.getTbNameFn = stmtGetTbName,
.setInfoFn = stmtUpdateInfo,
.getExecInfoFn = stmtGetExecInfo,
};
if (NULL == pStmt->exec.pRequest) {
......@@ -259,12 +259,12 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
}
size_t keyLen = 0;
void *pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
void* pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
while (pIter) {
STableDataBlocks* pBlocks = *(STableDataBlocks**)pIter;
char *key = taosHashGetKey(pIter, &keyLen);
STableMeta* pMeta = qGetTableMetaInDataBlock(pBlocks);
STableDataBlocks* pBlocks = *(STableDataBlocks**)pIter;
char* key = taosHashGetKey(pIter, &keyLen);
STableMeta* pMeta = qGetTableMetaInDataBlock(pBlocks);
if (keepTable && (strlen(pStmt->bInfo.tbFName) == keyLen) && strncmp(pStmt->bInfo.tbFName, key, keyLen) == 0) {
STMT_ERR_RET(qResetStmtDataBlock(pBlocks, true));
......@@ -279,7 +279,7 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
}
pStmt->exec.autoCreateTbl = false;
if (keepTable) {
return TSDB_CODE_SUCCESS;
}
......@@ -320,13 +320,15 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS;
}
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks *pDataBlock, STableDataBlocks **newBlock, uint64_t uid) {
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STableDataBlocks** newBlock, uint64_t uid) {
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
SVgroupInfo vgInfo = {0};
STMT_ERR_RET(catalogGetTableHashVgroup(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &vgInfo));
STMT_ERR_RET(taosHashPut(pStmt->exec.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo)));
STMT_ERR_RET(catalogGetTableHashVgroup(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname,
&vgInfo));
STMT_ERR_RET(
taosHashPut(pStmt->exec.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo)));
STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, vgInfo.vgId));
return TSDB_CODE_SUCCESS;
......@@ -335,8 +337,9 @@ int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks *pDataBlock, STab
int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->bInfo.needParse = true;
pStmt->bInfo.inExecCache = false;
STableDataBlocks *pBlockInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataBlocks* pBlockInExec =
taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (pBlockInExec) {
pStmt->bInfo.needParse = false;
pStmt->bInfo.inExecCache = true;
......@@ -352,7 +355,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->bInfo.needParse = false;
return TSDB_CODE_SUCCESS;
}
return TSDB_CODE_SUCCESS;
}
......@@ -367,24 +370,25 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->exec.autoCreateTbl = true;
pStmt->bInfo.tbUid = 0;
STableDataBlocks* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, 0));
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock, POINTER_BYTES)) {
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
POINTER_BYTES)) {
STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
return TSDB_CODE_SUCCESS;
}
STMT_RET(stmtCleanBindInfo(pStmt));
}
STableMeta *pTableMeta = NULL;
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
int32_t code = catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta);
STableMeta* pTableMeta = NULL;
SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
int32_t code =
catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta);
if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
STMT_ERR_RET(stmtCleanBindInfo(pStmt));
......@@ -398,7 +402,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
int8_t tableType = pTableMeta->tableType;
taosMemoryFree(pTableMeta);
uint64_t cacheUid = (TSDB_CHILD_TABLE == tableType) ? suid : uid;
if (uid == pStmt->bInfo.tbUid) {
pStmt->bInfo.needParse = false;
......@@ -408,8 +412,9 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
if (pStmt->bInfo.inExecCache) {
SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &cacheUid, sizeof(cacheUid));
if (NULL == pCache) {
tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash", pStmt->bInfo.tbFName, uid, cacheUid);
tscError("table [%s, %" PRIx64 ", %" PRIx64 "] found in exec blockHash, but not in sql blockHash",
pStmt->bInfo.tbFName, uid, cacheUid);
STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
}
......@@ -437,7 +442,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
STableDataBlocks* pNewBlock = NULL;
STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, uid));
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock, POINTER_BYTES)) {
if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
POINTER_BYTES)) {
STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
}
......@@ -521,10 +527,11 @@ int stmtSetTbName(TAOS_STMT* stmt, const char* tbName) {
if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
}
STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb,
pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
tNameExtractFullName(&pStmt->bInfo.sname, pStmt->bInfo.tbFName);
STMT_ERR_RET(stmtGetFromCache(pStmt));
if (pStmt->bInfo.needParse) {
......@@ -548,7 +555,8 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
return TSDB_CODE_SUCCESS;
}
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
......@@ -566,7 +574,8 @@ int32_t stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fiel
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
......@@ -583,7 +592,8 @@ int32_t stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD** fiel
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
......@@ -618,8 +628,8 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
}
if (STMT_TYPE_QUERY == pStmt->sql.type) {
STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx, pStmt->exec.pRequest->requestId));
STMT_ERR_RET(qStmtBindParams(pStmt->sql.pQuery, bind, colIdx));
SParseContext ctx = {.requestId = pStmt->exec.pRequest->requestId,
.acctId = pStmt->taos->acctId,
.db = pStmt->exec.pRequest->pDb,
......@@ -633,27 +643,29 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
.pUser = pStmt->taos->user};
ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
STMT_ERR_RET(qStmtParseQuerySql(&ctx, pStmt->sql.pQuery));
if (pStmt->sql.pQuery->haveResultSet) {
setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema, pStmt->sql.pQuery->numOfResCols);
setResSchemaInfo(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->pResSchema,
pStmt->sql.pQuery->numOfResCols);
setResPrecision(&pStmt->exec.pRequest->body.resInfo, pStmt->sql.pQuery->precision);
}
TSWAP(pStmt->exec.pRequest->dbList, pStmt->sql.pQuery->pDbList);
TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
TSWAP(pStmt->exec.pRequest->tableList, pStmt->sql.pQuery->pTableList);
//if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
// STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
//}
// if (STMT_TYPE_QUERY == pStmt->sql.queryRes) {
// STMT_ERR_RET(stmtRestoreQueryFields(pStmt));
// }
//STMT_ERR_RET(stmtBackupQueryFields(pStmt));
// STMT_ERR_RET(stmtBackupQueryFields(pStmt));
return TSDB_CODE_SUCCESS;
}
STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
STableDataBlocks** pDataBlock =
(STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
......@@ -694,19 +706,19 @@ int stmtAddBatch(TAOS_STMT* stmt) {
return TSDB_CODE_SUCCESS;
}
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp *pRsp) {
int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
if (pRsp->nBlocks <= 0) {
tscError("invalid submit resp block number %d", pRsp->nBlocks);
STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
}
size_t keyLen = 0;
STableDataBlocks **pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
size_t keyLen = 0;
STableDataBlocks** pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
while (pIter) {
STableDataBlocks *pBlock = *pIter;
char *key = taosHashGetKey(pIter, &keyLen);
STableMeta *pMeta = qGetTableMetaInDataBlock(pBlock);
STableDataBlocks* pBlock = *pIter;
char* key = taosHashGetKey(pIter, &keyLen);
STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
if (pMeta->uid != pStmt->bInfo.tbUid) {
tscError("table uid %" PRIx64 " mis-match with current table uid %" PRIx64, pMeta->uid, pStmt->bInfo.tbUid);
STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
......@@ -717,24 +729,25 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp *pRsp) {
continue;
}
SSubmitBlkRsp *blkRsp = NULL;
int32_t i = 0;
SSubmitBlkRsp* blkRsp = NULL;
int32_t i = 0;
for (; i < pRsp->nBlocks; ++i) {
blkRsp = pRsp->pBlocks + i;
if (strlen(blkRsp->tblFName) != keyLen) {
continue;
}
if (strncmp(blkRsp->tblFName, key, keyLen)) {
continue;
}
break;
}
if (i < pRsp->nBlocks) {
tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid, blkRsp->uid);
tscDebug("auto created table %s uid updated from %" PRIx64 " to %" PRIx64, blkRsp->tblFName, pMeta->uid,
blkRsp->uid);
pMeta->uid = blkRsp->uid;
pStmt->bInfo.tbUid = blkRsp->uid;
} else {
......@@ -748,11 +761,11 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp *pRsp) {
return TSDB_CODE_SUCCESS;
}
int stmtExec(TAOS_STMT *stmt) {
STscStmt* pStmt = (STscStmt*)stmt;
int32_t code = 0;
SSubmitRsp *pRsp = NULL;
bool autoCreateTbl = pStmt->exec.autoCreateTbl;
int stmtExec(TAOS_STMT* stmt) {
STscStmt* pStmt = (STscStmt*)stmt;
int32_t code = 0;
SSubmitRsp* pRsp = NULL;
bool autoCreateTbl = pStmt->exec.autoCreateTbl;
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
......@@ -760,7 +773,8 @@ int stmtExec(TAOS_STMT *stmt) {
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, NULL);
} else {
STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true, (autoCreateTbl ? (void**)&pRsp : NULL));
launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true,
(autoCreateTbl ? (void**)&pRsp : NULL));
}
if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
......@@ -787,10 +801,10 @@ _return:
tscError("no submit resp got for auto create table");
STMT_ERR_RET(TSDB_CODE_TSC_APP_ERROR);
}
STMT_ERR_RET(stmtUpdateTableUid(pStmt, pRsp));
}
++pStmt->sql.runTimes;
STMT_RET(code);
......
source/common/src/tglobal.c
浏览文件 @ e515a4ab
......@@ -433,7 +433,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
tsRpcQueueMemoryAllowed = tsTotalMemoryKB * 1024 * 0.1;
tsRpcQueueMemoryAllowed = TRANGE(tsRpcQueueMemoryAllowed, TSDB_MAX_WAL_SIZE * 10L, TSDB_MAX_WAL_SIZE * 10000L);
if (cfgAddInt64(pCfg, "rpcQueueMemoryAllowed", tsRpcQueueMemoryAllowed, 1, INT64_MAX, 0) != 0) return -1;
if (cfgAddInt64(pCfg, "rpcQueueMemoryAllowed", tsRpcQueueMemoryAllowed, TSDB_MAX_WAL_SIZE * 10L, INT64_MAX, 0) != 0)
return -1;
if (cfgAddBool(pCfg, "monitor", tsEnableMonitor, 0) != 0) return -1;
if (cfgAddInt32(pCfg, "monitorInterval", tsMonitorInterval, 1, 200000, 0) != 0) return -1;
......
source/dnode/mgmt/test/sut/src/sut.cpp
浏览文件 @ e515a4ab
......@@ -30,6 +30,7 @@ void Testbase::InitLog(const char* path) {
tsdbDebugFlag = 0;
tsLogEmbedded = 1;
tsAsyncLog = 0;
tsRpcQueueMemoryAllowed = 1024 * 1024 * 64;
taosRemoveDir(path);
taosMkDir(path);
......@@ -82,7 +83,7 @@ SRpcMsg* Testbase::SendReq(tmsg_t msgType, void* pCont, int32_t contLen) {
return client.SendReq(&rpcMsg);
}
int32_t Testbase::SendShowReq(int8_t showType, const char *tb, const char* db) {
int32_t Testbase::SendShowReq(int8_t showType, const char* tb, const char* db) {
if (showRsp != NULL) {
rpcFreeCont(showRsp);
showRsp = NULL;
......
source/libs/nodes/src/nodesUtilFuncs.c
浏览文件 @ e515a4ab
......@@ -529,6 +529,18 @@ void nodesDestroyNode(SNodeptr pNode) {
nodesDestroyNode(pStmt->pTbNamePattern);
break;
}
case QUERY_NODE_QUERY: {
SQuery* pQuery = (SQuery*)pNode;
nodesDestroyNode(pQuery->pRoot);
taosMemoryFreeClear(pQuery->pResSchema);
if (NULL != pQuery->pCmdMsg) {
taosMemoryFreeClear(pQuery->pCmdMsg->pMsg);
taosMemoryFreeClear(pQuery->pCmdMsg);
}
taosArrayDestroy(pQuery->pDbList);
taosArrayDestroy(pQuery->pTableList);
break;
}
case QUERY_NODE_LOGIC_PLAN_SCAN: {
SScanLogicNode* pLogicNode = (SScanLogicNode*)pNode;
destroyLogicNode((SLogicNode*)pLogicNode);
......
source/libs/parser/src/parTranslater.c
浏览文件 @ e515a4ab
......@@ -4254,7 +4254,135 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
return rewriteToVnodeModifyOpStmt(pQuery, pBufArray);
}
static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq) {
static SSchema* getColSchema(STableMeta* pTableMeta, const char* pTagName) {
int32_t numOfFields = getNumOfTags(pTableMeta) + getNumOfColumns(pTableMeta);
for (int32_t i = 0; i < numOfFields; ++i) {
SSchema* pTagSchema = pTableMeta->schema + i;
if (0 == strcmp(pTagName, pTagSchema->name)) {
return pTagSchema;
}
}
return NULL;
}
static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
}
pReq->tagName = strdup(pStmt->colName);
if (NULL == pReq->tagName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (DEAL_RES_ERROR == translateValueImpl(pCxt, pStmt->pVal, schemaToDataType(pSchema))) {
return pCxt->errCode;
}
pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
pReq->nTagVal = pStmt->pVal->node.resType.bytes;
char* pVal = nodesGetValueFromNode(pStmt->pVal);
pReq->pTagVal = IS_VAR_DATA_TYPE(pStmt->pVal->node.resType.type) ? pVal + VARSTR_HEADER_SIZE : pVal;
return TSDB_CODE_SUCCESS;
}
static int32_t buildAddColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
if (NULL != getColSchema(pTableMeta, pStmt->colName)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_DUPLICATED_COLUMN);
}
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pReq->type = pStmt->dataType.type;
pReq->flags = COL_SMA_ON;
pReq->bytes = pStmt->dataType.bytes;
return TSDB_CODE_SUCCESS;
}
static int32_t buildDropColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
} else if (PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY);
}
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildUpdateColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
pReq->colModBytes = calcTypeBytes(pStmt->dataType);
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
} else if (!IS_VAR_DATA_TYPE(pSchema->type) || pSchema->bytes >= pReq->colModBytes) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_MODIFY_COL);
}
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildRenameColReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
if (NULL == getColSchema(pTableMeta, pStmt->colName)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_COLUMN, pStmt->colName);
}
if (NULL != getColSchema(pTableMeta, pStmt->newColName)) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_DUPLICATED_COLUMN);
}
pReq->colName = strdup(pStmt->colName);
pReq->colNewName = strdup(pStmt->newColName);
if (NULL == pReq->colName || NULL == pReq->colNewName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
static int32_t buildUpdateOptionsReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq) {
int32_t code = TSDB_CODE_SUCCESS;
if (-1 != pStmt->pOptions->ttl) {
code = checkRangeOption(pCxt, "ttl", pStmt->pOptions->ttl, TSDB_MIN_TABLE_TTL, INT32_MAX);
if (TSDB_CODE_SUCCESS == code) {
pReq->updateTTL = true;
pReq->newTTL = pStmt->pOptions->ttl;
}
}
if (TSDB_CODE_SUCCESS == code && '\0' != pStmt->pOptions->comment[0]) {
pReq->updateComment = true;
pReq->newComment = strdup(pStmt->pOptions->comment);
if (NULL == pReq->newComment) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
return code;
}
static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
pReq->tbName = strdup(pStmt->tableName);
if (NULL == pReq->tbName) {
return TSDB_CODE_OUT_OF_MEMORY;
......@@ -4268,60 +4396,22 @@ static int32_t buildAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt,
case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
case TSDB_ALTER_TABLE_UPDATE_TAG_VAL:
pReq->tagName = strdup(pStmt->colName);
if (NULL == pReq->tagName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
if (DEAL_RES_ERROR == translateValue(pCxt, pStmt->pVal)) {
return pCxt->errCode;
}
pReq->isNull = (TSDB_DATA_TYPE_NULL == pStmt->pVal->node.resType.type);
pReq->nTagVal = pStmt->pVal->node.resType.bytes;
char* pVal = nodesGetValueFromNode(pStmt->pVal);
pReq->pTagVal = IS_VAR_DATA_TYPE(pStmt->pVal->node.resType.type) ? pVal + VARSTR_HEADER_SIZE : pVal;
break;
return buildUpdateTagValReq(pCxt, pStmt, pTableMeta, pReq);
case TSDB_ALTER_TABLE_ADD_COLUMN:
return buildAddColReq(pCxt, pStmt, pTableMeta, pReq);
case TSDB_ALTER_TABLE_DROP_COLUMN:
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pReq->type = pStmt->dataType.type;
pReq->flags = COL_SMA_ON;
pReq->bytes = pStmt->dataType.bytes;
break;
return buildDropColReq(pCxt, pStmt, pTableMeta, pReq);
case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
pReq->colName = strdup(pStmt->colName);
if (NULL == pReq->colName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
pReq->colModBytes = calcTypeBytes(pStmt->dataType);
break;
return buildUpdateColReq(pCxt, pStmt, pTableMeta, pReq);
case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
if (-1 != pStmt->pOptions->ttl) {
pReq->updateTTL = true;
pReq->newTTL = pStmt->pOptions->ttl;
}
if ('\0' != pStmt->pOptions->comment[0]) {
pReq->updateComment = true;
pReq->newComment = strdup(pStmt->pOptions->comment);
if (NULL == pReq->newComment) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
break;
return buildUpdateOptionsReq(pCxt, pStmt, pReq);
case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME:
pReq->colName = strdup(pStmt->colName);
pReq->colNewName = strdup(pStmt->newColName);
if (NULL == pReq->colName || NULL == pReq->colNewName) {
return TSDB_CODE_OUT_OF_MEMORY;
}
break;
return buildRenameColReq(pCxt, pStmt, pTableMeta, pReq);
default:
break;
}
return TSDB_CODE_SUCCESS;
return TSDB_CODE_FAILED;
}
static int32_t serializeAlterTbReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, SVAlterTbReq* pReq,
......@@ -4394,7 +4484,7 @@ static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
}
SVAlterTbReq req = {0};
code = buildAlterTbReq(pCxt, pStmt, &req);
code = buildAlterTbReq(pCxt, pStmt, pTableMeta, &req);
SArray* pArray = NULL;
if (TSDB_CODE_SUCCESS == code) {
......
source/libs/parser/src/parUtil.c
浏览文件 @ e515a4ab
......@@ -154,6 +154,10 @@ static char* getSyntaxErrFormat(int32_t errCode) {
return "Invalid password";
case TSDB_CODE_PAR_INVALID_ALTER_TABLE:
return "Invalid alter table statement";
case TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY:
return "Primary timestamp column cannot be dropped";
case TSDB_CODE_PAR_INVALID_MODIFY_COL:
return "Only binary/nchar column length could be modified";
case TSDB_CODE_OUT_OF_MEMORY:
return "Out of memory";
default:
......
source/libs/parser/src/parser.c
浏览文件 @ e515a4ab
......@@ -39,10 +39,16 @@ static int32_t parseSqlIntoAst(SParseContext* pCxt, SQuery** pQuery) {
if (TSDB_CODE_SUCCESS == code) {
code = authenticate(pCxt, *pQuery);
}
if (TSDB_CODE_SUCCESS == code && 0 == (*pQuery)->placeholderNum) {
if (TSDB_CODE_SUCCESS == code && (*pQuery)->placeholderNum > 0) {
// TSWAP((*pQuery)->pContainPlaceholderRoot, (*pQuery)->pRoot);
return TSDB_CODE_SUCCESS;
}
if (TSDB_CODE_SUCCESS == code) {
code = translate(pCxt, *pQuery);
}
if (TSDB_CODE_SUCCESS == code && 0 == (*pQuery)->placeholderNum) {
if (TSDB_CODE_SUCCESS == code) {
code = calculateConstant(pCxt, *pQuery);
}
return code;
......@@ -142,26 +148,13 @@ int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery) {
return code;
}
void qDestroyQuery(SQuery* pQueryNode) {
if (NULL == pQueryNode) {
return;
}
nodesDestroyNode(pQueryNode->pRoot);
taosMemoryFreeClear(pQueryNode->pResSchema);
if (NULL != pQueryNode->pCmdMsg) {
taosMemoryFreeClear(pQueryNode->pCmdMsg->pMsg);
taosMemoryFreeClear(pQueryNode->pCmdMsg);
}
taosArrayDestroy(pQueryNode->pDbList);
taosArrayDestroy(pQueryNode->pTableList);
taosMemoryFreeClear(pQueryNode);
}
void qDestroyQuery(SQuery* pQueryNode) { nodesDestroyNode(pQueryNode); }
int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {
return extractResultSchema(pRoot, numOfCols, pSchema);
}
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId) {
int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) {
int32_t code = TSDB_CODE_SUCCESS;
if (colIdx < 0) {
......@@ -184,6 +177,6 @@ int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery) {
if (TSDB_CODE_SUCCESS == code) {
code = calculateConstant(pCxt, pQuery);
}
return code;
}
source/libs/parser/test/parInitialATest.cpp
浏览文件 @ e515a4ab
......@@ -283,13 +283,13 @@ TEST_F(ParserInitialATest, alterTable) {
setAlterColFunc("t1", TSDB_ALTER_TABLE_DROP_COLUMN, "c1");
run("ALTER TABLE t1 DROP COLUMN c1");
setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, "c1", TSDB_DATA_TYPE_VARCHAR, 20 + VARSTR_HEADER_SIZE);
run("ALTER TABLE t1 MODIFY COLUMN c1 VARCHAR(20)");
setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES, "c2", TSDB_DATA_TYPE_VARCHAR, 30 + VARSTR_HEADER_SIZE);
run("ALTER TABLE t1 MODIFY COLUMN c2 VARCHAR(30)");
setAlterColFunc("t1", TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME, "c1", 0, 0, "cc1");
run("ALTER TABLE t1 RENAME COLUMN c1 cc1");
int64_t val = 10;
int32_t val = 10;
setAlterTagFunc("st1s1", "tag1", (const uint8_t*)&val, sizeof(val));
run("ALTER TABLE st1s1 SET TAG tag1=10");
......
source/libs/planner/src/planner.c
浏览文件 @ e515a4ab
......@@ -18,28 +18,6 @@
#include "planInt.h"
#include "scalar.h"
typedef struct SCollectPlaceholderValuesCxt {
int32_t errCode;
SArray* pValues;
} SCollectPlaceholderValuesCxt;
static EDealRes collectPlaceholderValuesImpl(SNode* pNode, void* pContext) {
if (QUERY_NODE_VALUE == nodeType(pNode) && ((SValueNode*)pNode)->placeholderNo > 0) {
SCollectPlaceholderValuesCxt* pCxt = pContext;
taosArrayInsert(pCxt->pValues, ((SValueNode*)pNode)->placeholderNo - 1, &pNode);
return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_IGNORE_CHILD : DEAL_RES_ERROR;
}
return DEAL_RES_CONTINUE;
}
static int32_t collectPlaceholderValues(SPlanContext* pCxt, SQueryPlan* pPlan) {
pPlan->pPlaceholderValues = taosArrayInit(TARRAY_MIN_SIZE, POINTER_BYTES);
SCollectPlaceholderValuesCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pValues = pPlan->pPlaceholderValues};
nodesWalkPhysiPlan((SNode*)pPlan, collectPlaceholderValuesImpl, &cxt);
return cxt.errCode;
}
int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNodeList) {
SLogicNode* pLogicNode = NULL;
SLogicSubplan* pLogicSubplan = NULL;
......@@ -58,9 +36,6 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
if (TSDB_CODE_SUCCESS == code) {
code = createPhysiPlan(pCxt, pLogicPlan, pPlan, pExecNodeList);
}
if (TSDB_CODE_SUCCESS == code && pCxt->placeholderNum > 0) {
code = collectPlaceholderValues(pCxt, *pPlan);
}
nodesDestroyNode(pLogicNode);
nodesDestroyNode(pLogicSubplan);
......@@ -99,249 +74,6 @@ int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstream
return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
}
static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
if (pParam->is_null && 1 == *(pParam->is_null)) {
pVal->node.resType.type = TSDB_DATA_TYPE_NULL;
pVal->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_NULL].bytes;
return TSDB_CODE_SUCCESS;
}
int32_t inputSize = (NULL != pParam->length ? *(pParam->length) : tDataTypes[pParam->buffer_type].bytes);
pVal->node.resType.type = pParam->buffer_type;
pVal->node.resType.bytes = inputSize;
switch (pParam->buffer_type) {
case TSDB_DATA_TYPE_BOOL:
pVal->datum.b = *((bool*)pParam->buffer);
break;
case TSDB_DATA_TYPE_TINYINT:
pVal->datum.i = *((int8_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_SMALLINT:
pVal->datum.i = *((int16_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_INT:
pVal->datum.i = *((int32_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_BIGINT:
pVal->datum.i = *((int64_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_FLOAT:
pVal->datum.d = *((float*)pParam->buffer);
break;
case TSDB_DATA_TYPE_DOUBLE:
pVal->datum.d = *((double*)pParam->buffer);
break;
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_VARBINARY:
pVal->datum.p = taosMemoryCalloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
if (NULL == pVal->datum.p) {
return TSDB_CODE_OUT_OF_MEMORY;
}
varDataSetLen(pVal->datum.p, pVal->node.resType.bytes);
strncpy(varDataVal(pVal->datum.p), (const char*)pParam->buffer, pVal->node.resType.bytes);
break;
case TSDB_DATA_TYPE_NCHAR: {
pVal->node.resType.bytes *= TSDB_NCHAR_SIZE;
pVal->datum.p = taosMemoryCalloc(1, pVal->node.resType.bytes + VARSTR_HEADER_SIZE + 1);
if (NULL == pVal->datum.p) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t output = 0;
if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes,
&output)) {
return errno;
}
varDataSetLen(pVal->datum.p, output);
pVal->node.resType.bytes = output;
break;
}
case TSDB_DATA_TYPE_TIMESTAMP:
pVal->datum.i = *((int64_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_UTINYINT:
pVal->datum.u = *((uint8_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_USMALLINT:
pVal->datum.u = *((uint16_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_UINT:
pVal->datum.u = *((uint32_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_UBIGINT:
pVal->datum.u = *((uint64_t*)pParam->buffer);
break;
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_MEDIUMBLOB:
// todo
default:
break;
}
pVal->translate = true;
return TSDB_CODE_SUCCESS;
}
static EDealRes updatePlanQueryId(SNode* pNode, void* pContext) {
int64_t queryId = *(uint64_t*)pContext;
if (QUERY_NODE_PHYSICAL_PLAN == nodeType(pNode)) {
SQueryPlan* planNode = (SQueryPlan*)pNode;
planNode->queryId = queryId;
} else if (QUERY_NODE_PHYSICAL_SUBPLAN == nodeType(pNode)) {
SSubplan* subplanNode = (SSubplan*)pNode;
subplanNode->id.queryId = queryId;
}
return DEAL_RES_CONTINUE;
}
static int32_t calcConstNode(SNode** pNode) {
if (NULL == *pNode) {
return TSDB_CODE_SUCCESS;
}
SNode* pNew = NULL;
int32_t code = scalarCalculateConstants(*pNode, &pNew);
if (TSDB_CODE_SUCCESS == code) {
*pNode = pNew;
}
return code;
}
static int32_t calcConstList(SNodeList* pList) {
SNode* pNode = NULL;
FOREACH(pNode, pList) {
SNode* pNew = NULL;
int32_t code = scalarCalculateConstants(pNode, &pNew);
if (TSDB_CODE_SUCCESS == code) {
REPLACE_NODE(pNew);
} else {
return code;
}
}
return TSDB_CODE_SUCCESS;
}
static bool isEmptyResultCond(SNode** pCond) {
if (NULL == *pCond || QUERY_NODE_VALUE != nodeType(*pCond)) {
return false;
}
if (((SValueNode*)*pCond)->datum.b) {
nodesDestroyNode(*pCond);
*pCond = NULL;
return false;
}
return true;
}
static int32_t calcConstSpecificPhysiNode(SPhysiNode* pPhyNode) {
switch (nodeType(pPhyNode)) {
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
case QUERY_NODE_PHYSICAL_PLAN_FILL:
return TSDB_CODE_SUCCESS;
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return calcConstList(((SProjectPhysiNode*)pPhyNode)->pProjections);
case QUERY_NODE_PHYSICAL_PLAN_JOIN:
return calcConstNode(&(((SJoinPhysiNode*)pPhyNode)->pOnConditions));
case QUERY_NODE_PHYSICAL_PLAN_AGG:
return calcConstList(((SAggPhysiNode*)pPhyNode)->pExprs);
case QUERY_NODE_PHYSICAL_PLAN_SORT:
return calcConstList(((SSortPhysiNode*)pPhyNode)->pExprs);
case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
return calcConstList(((SWinodwPhysiNode*)pPhyNode)->pExprs);
case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
return calcConstList(((SPartitionPhysiNode*)pPhyNode)->pExprs);
default:
break;
}
return TSDB_CODE_SUCCESS;
}
static int32_t calcConstSubplan(SPhysiNode* pPhyNode, bool* pEmptyResult) {
int32_t code = calcConstNode(&pPhyNode->pConditions);
if (TSDB_CODE_SUCCESS == code) {
code = calcConstSpecificPhysiNode(pPhyNode);
}
if (TSDB_CODE_SUCCESS != code) {
return code;
}
*pEmptyResult = isEmptyResultCond(&pPhyNode->pConditions);
if (*pEmptyResult) {
return TSDB_CODE_SUCCESS;
}
*pEmptyResult = true;
bool subEmptyResult = false;
SNode* pChild = NULL;
FOREACH(pChild, pPhyNode->pChildren) {
code = calcConstSubplan((SPhysiNode*)pChild, &subEmptyResult);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (!subEmptyResult) {
*pEmptyResult = false;
}
}
return TSDB_CODE_SUCCESS;
}
static int32_t calcConstPhysiPlan(SQueryPlan* pPlan, bool* pEmptyResult) {
*pEmptyResult = true;
bool subEmptyResult = false;
SNodeListNode* pNode = nodesListGetNode(pPlan->pSubplans, 0);
SNode* pSubplan = NULL;
FOREACH(pSubplan, pNode->pNodeList) {
int32_t code = calcConstSubplan(((SSubplan*)pSubplan)->pNode, pEmptyResult);
if (TSDB_CODE_SUCCESS != code) {
return code;
}
if (!subEmptyResult) {
*pEmptyResult = false;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId,
bool* pEmptyResult) {
int32_t size = taosArrayGetSize(pPlan->pPlaceholderValues);
int32_t code = 0;
if (colIdx < 0) {
for (int32_t i = 0; i < size; ++i) {
code = setValueByBindParam((SValueNode*)taosArrayGetP(pPlan->pPlaceholderValues, i), pParams + i);
if (code) {
return code;
}
}
} else {
code = setValueByBindParam((SValueNode*)taosArrayGetP(pPlan->pPlaceholderValues, colIdx), pParams);
if (code) {
return code;
}
}
if (colIdx < 0 || ((colIdx + 1) == size)) {
nodesWalkPhysiPlan((SNode*)pPlan, updatePlanQueryId, &queryId);
code = calcConstPhysiPlan(pPlan, pEmptyResult);
}
return code;
}
int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {
if (SUBPLAN_TYPE_MODIFY == pSubplan->subplanType) {
SDataInserterNode* insert = (SDataInserterNode*)pSubplan->pDataSink;
......
source/libs/planner/test/planStmtTest.cpp
浏览文件 @ e515a4ab
......@@ -20,35 +20,49 @@ using namespace std;
class PlanStmtTest : public PlannerTestBase {
public:
void prepare(const string& sql) {
run(sql);
// todo calloc pBindParams_
}
void bindParam(int32_t val) {
TAOS_MULTI_BIND* pBind = pBindParams_ + paramNo_++;
pBind->buffer_type = TSDB_DATA_TYPE_INT;
pBind->num = 1;
pBind->buffer_length = sizeof(int32_t);
pBind->buffer = taosMemoryCalloc(1, pBind->buffer_length);
pBind->length = (int32_t*)taosMemoryCalloc(1, sizeof(int32_t));
pBind->is_null = (char*)taosMemoryCalloc(1, sizeof(char));
*((int32_t*)pBind->buffer) = val;
*(pBind->length) = sizeof(int32_t);
*(pBind->is_null) = 0;
}
void buildParam(TAOS_MULTI_BIND* pBindParams, int32_t index, void* pVal, int32_t type, int32_t bytes = 0) {
TAOS_MULTI_BIND* pBindParam = pBindParams + index;
pBindParam->buffer_type = type;
pBindParam->num = 1;
pBindParam->buffer_length = bytes > 0 ? bytes : tDataTypes[type].bytes;
pBindParam->buffer = taosMemoryCalloc(1, pBindParam->buffer_length);
pBindParam->length = (int32_t*)taosMemoryCalloc(1, sizeof(int32_t));
pBindParam->is_null = (char*)taosMemoryCalloc(1, sizeof(char));
*(pBindParam->length) = bytes > 0 ? bytes : tDataTypes[type].bytes;
*(pBindParam->is_null) = 0;
void exec() {
// todo
switch (type) {
case TSDB_DATA_TYPE_BOOL:
*((bool*)pBindParam->buffer) = *(bool*)pVal;
break;
case TSDB_DATA_TYPE_TINYINT:
*((int8_t*)pBindParam->buffer) = *(int64_t*)pVal;
break;
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_FLOAT:
case TSDB_DATA_TYPE_DOUBLE:
case TSDB_DATA_TYPE_VARCHAR:
case TSDB_DATA_TYPE_TIMESTAMP:
case TSDB_DATA_TYPE_NCHAR:
case TSDB_DATA_TYPE_UTINYINT:
case TSDB_DATA_TYPE_USMALLINT:
case TSDB_DATA_TYPE_UINT:
case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_MEDIUMBLOB:
default:
break;
}
}
private:
TAOS_MULTI_BIND* pBindParams_;
int32_t paramNo_;
};
TEST_F(PlanStmtTest, stmt) {
useDb("root", "test");
// run("select * from t1 where c1 = ?");
prepare("SELECT * FROM t1 WHERE c1 = ?");
}
source/libs/planner/test/planTestUtil.cpp
浏览文件 @ e515a4ab
......@@ -108,6 +108,57 @@ class PlannerTestBaseImpl {
}
}
void prepare(const string& sql) {
reset();
try {
doParseSql(sql, &stmtEnv_.pQuery_, true);
dump(g_dumpModule);
} catch (...) {
dump(DUMP_MODULE_ALL);
throw;
}
}
void bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx) {
try {
doBindParams(stmtEnv_.pQuery_, pParams, colIdx);
SPlanContext cxt = {0};
setPlanContext(stmtEnv_.pQuery_, &cxt);
SLogicNode* pLogicNode = nullptr;
doCreateLogicPlan(&cxt, &pLogicNode);
doOptimizeLogicPlan(&cxt, pLogicNode);
SLogicSubplan* pLogicSubplan = nullptr;
doSplitLogicPlan(&cxt, pLogicNode, &pLogicSubplan);
SQueryLogicPlan* pLogicPlan = nullptr;
doScaleOutLogicPlan(&cxt, pLogicSubplan, &pLogicPlan);
SQueryPlan* pPlan = nullptr;
doCreatePhysiPlan(&cxt, pLogicPlan, &pPlan);
dump(g_dumpModule);
} catch (...) {
dump(DUMP_MODULE_ALL);
throw;
}
}
void exec() {
try {
doParseBoundSql(stmtEnv_.pQuery_);
dump(g_dumpModule);
} catch (...) {
dump(DUMP_MODULE_ALL);
throw;
}
}
private:
struct caseEnv {
string acctId_;
......@@ -117,10 +168,15 @@ class PlannerTestBaseImpl {
struct stmtEnv {
string sql_;
array<char, 1024> msgBuf_;
SQuery* pQuery_;
~stmtEnv() { qDestroyQuery(pQuery_); }
};
struct stmtRes {
string ast_;
string prepareAst_;
string boundAst_;
string rawLogicPlan_;
string optimizedLogicPlan_;
string splitLogicPlan_;
......@@ -132,8 +188,10 @@ class PlannerTestBaseImpl {
void reset() {
stmtEnv_.sql_.clear();
stmtEnv_.msgBuf_.fill(0);
qDestroyQuery(stmtEnv_.pQuery_);
res_.ast_.clear();
res_.boundAst_.clear();
res_.rawLogicPlan_.clear();
res_.optimizedLogicPlan_.clear();
res_.splitLogicPlan_.clear();
......@@ -152,6 +210,9 @@ class PlannerTestBaseImpl {
if (DUMP_MODULE_ALL == module || DUMP_MODULE_PARSER == module) {
cout << "syntax tree : " << endl;
cout << res_.ast_ << endl;
cout << "bound syntax tree : " << endl;
cout << res_.boundAst_ << endl;
}
if (DUMP_MODULE_ALL == module || DUMP_MODULE_LOGIC == module) {
......@@ -187,7 +248,7 @@ class PlannerTestBaseImpl {
}
}
void doParseSql(const string& sql, SQuery** pQuery) {
void doParseSql(const string& sql, SQuery** pQuery, bool prepare = false) {
stmtEnv_.sql_ = sql;
transform(stmtEnv_.sql_.begin(), stmtEnv_.sql_.end(), stmtEnv_.sql_.begin(), ::tolower);
......@@ -200,7 +261,31 @@ class PlannerTestBaseImpl {
cxt.msgLen = stmtEnv_.msgBuf_.max_size();
DO_WITH_THROW(qParseSql, &cxt, pQuery);
res_.ast_ = toString((*pQuery)->pRoot);
if (prepare) {
res_.prepareAst_ = toString((*pQuery)->pRoot);
} else {
res_.ast_ = toString((*pQuery)->pRoot);
}
}
void doBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx) {
DO_WITH_THROW(qStmtBindParams, pQuery, pParams, colIdx);
if (colIdx < 0 || pQuery->placeholderNum == colIdx + 1) {
res_.boundAst_ = toString(pQuery->pRoot);
}
}
void doParseBoundSql(SQuery* pQuery) {
SParseContext cxt = {0};
cxt.acctId = atoi(caseEnv_.acctId_.c_str());
cxt.db = caseEnv_.db_.c_str();
cxt.pSql = stmtEnv_.sql_.c_str();
cxt.sqlLen = stmtEnv_.sql_.length();
cxt.pMsg = stmtEnv_.msgBuf_.data();
cxt.msgLen = stmtEnv_.msgBuf_.max_size();
DO_WITH_THROW(qStmtParseQuerySql, &cxt, pQuery);
res_.ast_ = toString(pQuery->pRoot);
}
void doCreateLogicPlan(SPlanContext* pCxt, SLogicNode** pLogicNode) {
......@@ -275,3 +360,11 @@ PlannerTestBase::~PlannerTestBase() {}
void PlannerTestBase::useDb(const std::string& acctId, const std::string& db) { impl_->useDb(acctId, db); }
void PlannerTestBase::run(const std::string& sql) { return impl_->run(sql); }
void PlannerTestBase::prepare(const std::string& sql) { return impl_->prepare(sql); }
void PlannerTestBase::bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx) {
return impl_->bindParams(pParams, colIdx);
}
void PlannerTestBase::exec() { return impl_->exec(); }
source/libs/planner/test/planTestUtil.h
浏览文件 @ e515a4ab
......@@ -19,6 +19,7 @@
#include <gtest/gtest.h>
class PlannerTestBaseImpl;
struct TAOS_MULTI_BIND;
class PlannerTestBase : public testing::Test {
public:
......@@ -27,6 +28,10 @@ class PlannerTestBase : public testing::Test {
void useDb(const std::string& acctId, const std::string& db);
void run(const std::string& sql);
// stmt mode APIs
void prepare(const std::string& sql);
void bindParams(TAOS_MULTI_BIND* pParams, int32_t colIdx);
void exec();
private:
std::unique_ptr<PlannerTestBaseImpl> impl_;
......
source/libs/scheduler/src/scheduler.c
浏览文件 @ e515a4ab
......@@ -1148,6 +1148,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
if (NULL == msg) {
SCH_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
}
SCH_ERR_RET(schProcessOnTaskSuccess(pJob, pTask));
break;
}
case TDMT_VND_SUBMIT_RSP: {
......
source/util/src/tqueue.c
浏览文件 @ e515a4ab
......@@ -33,11 +33,11 @@ typedef struct STaosQnode {
} STaosQnode;
typedef struct STaosQueue {
STaosQnode *head;
STaosQnode *tail;
STaosQueue *next; // for queue set
STaosQset *qset; // for queue set
void *ahandle; // for queue set
STaosQnode * head;
STaosQnode * tail;
STaosQueue * next; // for queue set
STaosQset * qset; // for queue set
void * ahandle; // for queue set
FItem itemFp;
FItems itemsFp;
TdThreadMutex mutex;
......@@ -46,8 +46,8 @@ typedef struct STaosQueue {
} STaosQueue;
typedef struct STaosQset {
STaosQueue *head;
STaosQueue *current;
STaosQueue * head;
STaosQueue * current;
TdThreadMutex mutex;
tsem_t sem;
int32_t numOfQueues;
......@@ -85,7 +85,7 @@ void taosSetQueueFp(STaosQueue *queue, FItem itemFp, FItems itemsFp) {
void taosCloseQueue(STaosQueue *queue) {
if (queue == NULL) return;
STaosQnode *pTemp;
STaosQset *qset;
STaosQset * qset;
taosThreadMutexLock(&queue->mutex);
STaosQnode *pNode = queue->head;
......@@ -152,7 +152,7 @@ void *taosAllocateQitem(int32_t size, EQItype itype) {
if (itype == RPC_QITEM) {
int64_t alloced = atomic_add_fetch_64(&tsRpcQueueMemoryUsed, size);
if (alloced > tsRpcQueueMemoryUsed) {
if (alloced > tsRpcQueueMemoryAllowed) {
taosMemoryFree(pNode);
terrno = TSDB_CODE_OUT_OF_RPC_MEMORY_QUEUE;
return NULL;
......
source/util/test/procTest.cpp
浏览文件 @ e515a4ab
......@@ -16,11 +16,11 @@
typedef struct STestMsg {
uint16_t msgType;
void *pCont;
void * pCont;
int contLen;
int32_t code;
void *handle; // rpc handle returned to app
void *ahandle; // app handle set by client
void * handle; // rpc handle returned to app
void * ahandle; // app handle set by client
int noResp; // has response or not(default 0, 0: resp, 1: no resp);
int persistHandle; // persist handle or not
} STestMsg;
......@@ -37,7 +37,7 @@ class UtilTesProc : public ::testing::Test {
head.msgType = 2;
head.noResp = 3;
head.persistHandle = 4;
tsRpcQueueMemoryAllowed = 1024 * 1024 * 64;
taosRemoveDir("/tmp/td");
taosMkDir("/tmp/td");
tstrncpy(tsLogDir, "/tmp/td", PATH_MAX);
......@@ -64,18 +64,18 @@ TEST_F(UtilTesProc, 00_Init_Cleanup) {
shm.size = 1023;
SProcCfg cfg = {(ProcConsumeFp)NULL,
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryMalloc,
(ProcConsumeFp)NULL,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryMalloc,
shm,
&shm,
"1234"};
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryMalloc,
(ProcConsumeFp)NULL,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryMalloc,
shm,
&shm,
"1234"};
SProcObj *proc = taosProcInit(&cfg);
ASSERT_EQ(proc, nullptr);
......@@ -105,18 +105,18 @@ TEST_F(UtilTesProc, 01_Push_Pop_Child) {
shm.size = 3000;
ASSERT_EQ(taosCreateShm(&shm, 1235, shm.size), 0);
SProcCfg cfg = {(ProcConsumeFp)ConsumeChild1,
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcConsumeFp)NULL,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
shm,
(void *)((int64_t)1235),
"1235_c"};
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcConsumeFp)NULL,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
shm,
(void *)((int64_t)1235),
"1235_c"};
SProcObj *cproc = taosProcInit(&cfg);
ASSERT_NE(cproc, nullptr);
......@@ -163,18 +163,18 @@ TEST_F(UtilTesProc, 02_Push_Pop_Parent) {
shm.size = 3000;
ASSERT_EQ(taosCreateShm(&shm, 1236, shm.size), 0);
SProcCfg cfg = {(ProcConsumeFp)NULL,
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcConsumeFp)ConsumeParent1,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
shm,
(void *)((int64_t)1236),
"1236_c"};
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcConsumeFp)ConsumeParent1,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
shm,
(void *)((int64_t)1236),
"1236_c"};
SProcObj *cproc = taosProcInit(&cfg);
ASSERT_NE(cproc, nullptr);
......@@ -217,18 +217,18 @@ TEST_F(UtilTesProc, 03_Handle) {
shm.size = 3000;
ASSERT_EQ(taosCreateShm(&shm, 1237, shm.size), 0);
SProcCfg cfg = {(ProcConsumeFp)ConsumeChild3,
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcConsumeFp)NULL,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
shm,
(void *)((int64_t)1235),
"1237_p"};
(ProcMallocFp)taosAllocateQitem,
(ProcFreeFp)taosFreeQitem,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcConsumeFp)NULL,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
(ProcMallocFp)taosMemoryMalloc,
(ProcFreeFp)taosMemoryFree,
shm,
(void *)((int64_t)1235),
"1237_p"};
SProcObj *cproc = taosProcInit(&cfg);
ASSERT_NE(cproc, nullptr);
......@@ -236,7 +236,8 @@ TEST_F(UtilTesProc, 03_Handle) {
int32_t i = 0;
for (i = 0; i < 20; ++i) {
head.handle = (void *)((int64_t)i);
ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, (void *)((int64_t)i), i, PROC_FUNC_REQ), 0);
ASSERT_EQ(taosProcPutToChildQ(cproc, &head, sizeof(STestMsg), body, i, (void *)((int64_t)i), i, PROC_FUNC_REQ),
0);
}
cfg.isChild = true;
......@@ -247,7 +248,7 @@ TEST_F(UtilTesProc, 03_Handle) {
taosProcCleanup(pproc);
int64_t ref = 0;
ref = taosProcRemoveHandle(cproc, (void *)((int64_t)3));
EXPECT_EQ(ref, 3);
ref = taosProcRemoveHandle(cproc, (void *)((int64_t)5));
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或