diff --git a/include/libs/nodes/querynodes.h b/include/libs/nodes/querynodes.h index 60f00daebe36b00da89f8543eaa985f35176dbb9..01f3132c69f3e25474b9f3fa74aa797e2fb529ef 100644 --- a/include/libs/nodes/querynodes.h +++ b/include/libs/nodes/querynodes.h @@ -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); diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 248953cbaf2468dff19e63e50ff8893cbba96d26..2d8fd9a93cadc1275e937ade9a8b859dcebe7dc9 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -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 diff --git a/include/libs/planner/planner.h b/include/libs/planner/planner.h index e250b7b2b2d8c66295047ca89eb9f4c0a9fe1ec5..c4f71e57a8174c62cf331e4afec35604786282a0 100644 --- a/include/libs/planner/planner.h +++ b/include/libs/planner/planner.h @@ -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); diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 31b6dad2d807d7be029bd4d5ccbf5ee6a276cf25..f2ab9cbc6e9496a852967acb9af869e0fb5781cd 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -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) diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 100ec42899ea88dcecc3aa6365507a8a4d0c53da..de08dbb34f9c367ea380f2091552ec7afb8ddf4a 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -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) { diff --git a/source/client/src/clientStmt.c b/source/client/src/clientStmt.c index 1e9cb7b24fa89d348d1b615960afa63277902a7a..fa8df2852339ff563ad85ba5df6da497f4fec3d0 100644 --- a/source/client/src/clientStmt.c +++ b/source/client/src/clientStmt.c @@ -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); diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index aaad606feb878b7e38f52c21fe97cc85434649bc..4920355a95e18529cd744aa03cd32053ce1f439c 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -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; diff --git a/source/dnode/mgmt/test/sut/src/sut.cpp b/source/dnode/mgmt/test/sut/src/sut.cpp index cc32f047afcd1baf9e3d0857fe68248187a29e0e..7d37422b5a251b82a0ea0f11dd1ef16ebf63bca4 100644 --- a/source/dnode/mgmt/test/sut/src/sut.cpp +++ b/source/dnode/mgmt/test/sut/src/sut.cpp @@ -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; diff --git a/source/libs/nodes/src/nodesUtilFuncs.c b/source/libs/nodes/src/nodesUtilFuncs.c index c2bde4082c129c8f40b46f87b9a5dd7b0ff472aa..76c8b1748722362b1ec1107a95627b6a93f1507f 100644 --- a/source/libs/nodes/src/nodesUtilFuncs.c +++ b/source/libs/nodes/src/nodesUtilFuncs.c @@ -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); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 3983b53b6e001cefaf13cce940989268abeb90c5..df7d10969d2dd7c2081b7e98292cd6476ba6cda6 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -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) { diff --git a/source/libs/parser/src/parUtil.c b/source/libs/parser/src/parUtil.c index e7716741ed5b7cd07fd01ca70c6fe0b59061d690..cb2d1b7c072683c9cd3f452b78c79e689b5dcacb 100644 --- a/source/libs/parser/src/parUtil.c +++ b/source/libs/parser/src/parUtil.c @@ -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: diff --git a/source/libs/parser/src/parser.c b/source/libs/parser/src/parser.c index 5962867869025190f6d8835f4dd558194d795fab..82d1aabd2abae1ed906b7c70296b22ccc924dfe2 100644 --- a/source/libs/parser/src/parser.c +++ b/source/libs/parser/src/parser.c @@ -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; } diff --git a/source/libs/parser/test/parInitialATest.cpp b/source/libs/parser/test/parInitialATest.cpp index e1fa8ffe8dd6a3cdd38b35edf3f0e231e3d929e9..cc0dded5701bbe72c062aa69d339454976e6d1ac 100644 --- a/source/libs/parser/test/parInitialATest.cpp +++ b/source/libs/parser/test/parInitialATest.cpp @@ -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"); diff --git a/source/libs/planner/src/planner.c b/source/libs/planner/src/planner.c index f65e674bf6c6e92dd9ceafd1112f98f3732b7f3e..8e6c04bb337b900b1ebc8b33c73b29d111231ab3 100644 --- a/source/libs/planner/src/planner.c +++ b/source/libs/planner/src/planner.c @@ -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; diff --git a/source/libs/planner/test/planStmtTest.cpp b/source/libs/planner/test/planStmtTest.cpp index 0674a69355e9e4706095aefb075a736f380e84e3..6d6eaaf190935ff8168ca6b5109db4956677116c 100644 --- a/source/libs/planner/test/planStmtTest.cpp +++ b/source/libs/planner/test/planStmtTest.cpp @@ -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 = ?"); } diff --git a/source/libs/planner/test/planTestUtil.cpp b/source/libs/planner/test/planTestUtil.cpp index 6b038ae8ea9b4f405928e93b313adb685d5b6aac..97ff181d7f57ae157eae0a0ab3634c0bea561650 100644 --- a/source/libs/planner/test/planTestUtil.cpp +++ b/source/libs/planner/test/planTestUtil.cpp @@ -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 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(); } diff --git a/source/libs/planner/test/planTestUtil.h b/source/libs/planner/test/planTestUtil.h index a63bba1a97738d4386f495f75dec78e0d1c1dd7c..5a2050a45e91d262751333e2f8dd12b02690af2e 100644 --- a/source/libs/planner/test/planTestUtil.h +++ b/source/libs/planner/test/planTestUtil.h @@ -19,6 +19,7 @@ #include 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 impl_; diff --git a/source/libs/scheduler/src/scheduler.c b/source/libs/scheduler/src/scheduler.c index 5637539feab58969af070ae812b589797800ce76..4fcdb35e0db5a7b958dcb3f576c75119f1c41b8c 100644 --- a/source/libs/scheduler/src/scheduler.c +++ b/source/libs/scheduler/src/scheduler.c @@ -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: { diff --git a/source/util/src/tqueue.c b/source/util/src/tqueue.c index 17a87b1a5fffa77c076c2201dcdd18312a09c488..5e206f3e6e0a55895089312c5172790713fee42a 100644 --- a/source/util/src/tqueue.c +++ b/source/util/src/tqueue.c @@ -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; diff --git a/source/util/test/procTest.cpp b/source/util/test/procTest.cpp index 7ffec04a40f97a878cdad262488f48a34125dc51..d77180138c6661d55c281154e130cad45197c729 100644 --- a/source/util/test/procTest.cpp +++ b/source/util/test/procTest.cpp @@ -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));