diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 84f1da167174..da8b7c5170e7 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -165,7 +165,18 @@ typedef struct SParseContext { void* charsetCxt; } SParseContext; +typedef struct SPureInsertParserCtx { + int nr_params; + + char buf[512]; +} SPureInsertParserCtx; + int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery); +bool qIsLiteralSql(const char* pStr); + +// NOTE: only for insert into [db.]? (...) values (...) +int32_t qPureParseInsert(SPureInsertParserCtx *pCtx, const char *pStr); + bool qIsInsertValuesSql(const char* pStr, size_t length); bool qIsUpdateSetSql(const char* pStr, size_t length, SName* pTableName, int32_t acctId, const char* dbName, char* msgBuf, int32_t msgBufLen, int* pCode); diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index 3cd9cb7cc047..8cfab62a7928 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -364,6 +364,8 @@ typedef struct SRequestObj { int32_t execPhase; // EQueryExecPhase int64_t phaseStartTime; // when current phase started, ms int8_t secureDelete; + + TAOS_STMT2 *literal_by_stmt2; // reference only } SRequestObj; typedef struct SSyncQueryParam { @@ -387,9 +389,11 @@ void syncCatalogFn(SMetaData* pResult, void* param, int32_t code); TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly, int8_t source); TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly, int64_t reqid); +void taosAsyncExecLiteral(TAOS_STMT2 *stmt); + void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, int8_t source); -void taosAsyncQueryImplWithReqid(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, +void taosAsyncQueryImplWithReqid(TAOS_STMT2 *stmt, uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, int64_t reqid); void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param); int32_t clientParseSql(void* param, const char* dbName, const char* sql, bool parseOnly, const char* effectiveUser, diff --git a/source/client/inc/clientStmt2.h b/source/client/inc/clientStmt2.h index aba1ab6b2c45..c91ac362f3d0 100644 --- a/source/client/inc/clientStmt2.h +++ b/source/client/inc/clientStmt2.h @@ -86,6 +86,18 @@ STableDataCxt *pCurrBlock; SSubmitTbData *pCurrTbData; } SStmtExecInfo; */ +typedef struct SStmt2LiteralCtx { + tsem_t sem; + + int32_t code; + + uint8_t sem_valid:1; // sem valid or not + uint8_t prepared:1; // literal statement prepared by stmt2 or not + uint8_t executing:1; // literal statement executing by stmt2 or not + uint8_t executed:1; // literal statement executed by stmt2 or not + uint8_t has_result_set:1; // literal statement generates result set or not +} SStmt2LiteralCtx; + typedef struct { bool stbInterlaceMode; STMT_TYPE type; @@ -176,6 +188,10 @@ typedef struct { bool asyncResultAvailable; SStmtStatInfo stat; SArray* pVgDataBlocksForRetry; // SArray saved serialized data for NEED_CLIENT_HANDLE_ERROR retry + + char msgBuf[128]; + SStmt2LiteralCtx ctx; + uint8_t literal:1; } STscStmt2; /* extern char *gStmtStatusStr[]; @@ -256,6 +272,7 @@ TAOS_STMT2 *stmtInit2(STscObj *taos, TAOS_STMT2_OPTION *pOptions); int stmtClose2(TAOS_STMT2 *stmt); int stmtExec2(TAOS_STMT2 *stmt, int *affected_rows); int stmtPrepare2(TAOS_STMT2 *stmt, const char *sql, unsigned long length); +int stmtBindLiteral2(TAOS_STMT2 *stmt); int stmtSetTbName2(TAOS_STMT2 *stmt, const char *tbName); int stmtSetTbTags2(TAOS_STMT2 *stmt, TAOS_STMT2_BIND *tags, SVCreateTbReq **pCreateTbReq); int stmtCheckTags2(TAOS_STMT2 *stmt, SVCreateTbReq **pCreateTbReq); @@ -271,6 +288,9 @@ int stmtAsyncBindThreadFunc(void *args); void stmtBuildErrorMsg(STscStmt2 *pStmt, const char *msg); int32_t stmtBuildErrorMsgWithCode(STscStmt2 *pStmt, const char *msg, int32_t errorCode); + +int stmtIsLiteral(TAOS_STMT2 *stmt); + #ifdef __cplusplus } #endif diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 9a6d56cd1fde..7f694d6ea400 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -3247,7 +3247,7 @@ void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, doAsyncQuery(pRequest, false); } -void taosAsyncQueryImplWithReqid(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, +void taosAsyncQueryImplWithReqid(TAOS_STMT2 *stmt, uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly, int64_t reqid) { if (sql == NULL || NULL == fp) { terrno = TSDB_CODE_INVALID_PARA; @@ -3276,6 +3276,8 @@ void taosAsyncQueryImplWithReqid(uint64_t connId, const char* sql, __taos_async_ return; } + pRequest->literal_by_stmt2 = stmt; + code = connCheckAndUpateMetric(connId); if (code != TSDB_CODE_SUCCESS) { @@ -3347,7 +3349,7 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly, return NULL; } - taosAsyncQueryImplWithReqid(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly, reqid); + taosAsyncQueryImplWithReqid(NULL, *(int64_t*)taos, sql, syncQueryFn, param, validateOnly, reqid); code = tsem_wait(¶m->sem); if (TSDB_CODE_SUCCESS != code) { taosMemoryFree(param); @@ -3460,6 +3462,13 @@ void doRequestCallback(SRequestObj* pRequest, int32_t code) { pRequest->inCallback = true; int64_t this = pRequest->self; + SRequestObj* pThis = acquireRequest(this); + if (pThis != pRequest) { + // NOTE: internal logic error, not recoverable!!! + tscError("internal logic error: SRequestObj lifecycle management"); + abort(); + } + if (tsQueryTbNotExistAsEmpty && TD_RES_QUERY(&pRequest->resType) && pRequest->isQuery && (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_TDB_TABLE_NOT_EXIST)) { code = TSDB_CODE_SUCCESS; @@ -3479,11 +3488,8 @@ void doRequestCallback(SRequestObj* pRequest, int32_t code) { pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code); } - SRequestObj* pReq = acquireRequest(this); - if (pReq != NULL) { - pReq->inCallback = false; - (void)releaseRequest(this); - } + pRequest->inCallback = false; + (void)releaseRequest(this); // NOTE: pairing `pThis = acquireRequest(this);` } int32_t clientParseSql(void* param, const char* dbName, const char* sql, bool parseOnly, const char* effectiveUser, diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index b6059a96837a..f3e87824b1c2 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -1897,7 +1897,8 @@ void handleQueryAnslyseRes(SSqlCallbackWrapper *pWrapper, SMetaData *pResultMeta qDestroyQuery(pRequest->pQuery); pRequest->pQuery = NULL; - if (NEED_CLIENT_HANDLE_ERROR(code) && pRequest->stmtBindVersion == 0) { + if (NEED_CLIENT_HANDLE_ERROR(code) && (pRequest->stmtBindVersion == 0 || (pRequest->stmtBindVersion == 2 && pRequest->literal_by_stmt2))) { + // NOTE: also cover literal statement by stmt2 tscDebug("req:0x%" PRIx64 ", client retry to handle the error, code:%d - %s, tryCount:%d, QID:0x%" PRIx64, pRequest->self, code, tstrerror(code), pRequest->retry, pRequest->requestId); restartAsyncQuery(pRequest, code); @@ -2005,7 +2006,7 @@ void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param void taos_query_a_with_reqid(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param, int64_t reqid) { int64_t connId = *(int64_t *)taos; - taosAsyncQueryImplWithReqid(connId, sql, fp, param, false, reqid); + taosAsyncQueryImplWithReqid(NULL, connId, sql, fp, param, false, reqid); } int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt, SSqlCallbackWrapper *pWrapper) { @@ -2089,24 +2090,8 @@ int32_t prepareAndParseSqlSyntax(SSqlCallbackWrapper **ppWrapper, SRequestObj *p return code; } -void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { - SSqlCallbackWrapper *pWrapper = NULL; - int32_t code = TSDB_CODE_SUCCESS; - - CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_PARSE); - - if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) { - code = pRequest->prevCode; - terrno = code; - pRequest->code = code; - tscDebug("req:0x%" PRIx64 ", call sync query cb with code:%s", pRequest->self, tstrerror(code)); - doRequestCallback(pRequest, code); - return; - } - - if (TSDB_CODE_SUCCESS == code) { - code = prepareAndParseSqlSyntax(&pWrapper, pRequest, updateMetaForce); - } +static void doAsyncExec(SRequestObj *pRequest, int32_t code) { + SSqlCallbackWrapper *pWrapper = pRequest->pWrapper; if (TSDB_CODE_SUCCESS == code) { pRequest->stmtType = pRequest->pQuery->pRoot->type; @@ -2146,6 +2131,44 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { } } +void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { + SSqlCallbackWrapper *pWrapper = NULL; + int32_t code = TSDB_CODE_SUCCESS; + + CLIENT_UPDATE_REQUEST_PHASE_IF_CHANGED(pRequest, QUERY_PHASE_PARSE); + + if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) { + code = pRequest->prevCode; + terrno = code; + pRequest->code = code; + tscDebug("req:0x%" PRIx64 ", call sync query cb with code:%s", pRequest->self, tstrerror(code)); + doRequestCallback(pRequest, code); + return; + } + + if (TSDB_CODE_SUCCESS == code) { + code = prepareAndParseSqlSyntax(&pWrapper, pRequest, updateMetaForce); + } + + if (pRequest->literal_by_stmt2) { + TAOS_STMT2* stmt = pRequest->literal_by_stmt2; + STscStmt2* pStmt = (STscStmt2*)stmt; + if (pStmt->ctx.prepared == 0) { + // NOTE: preparing stage for literal statement by stmt2 + doRequestCallback(pRequest, code); + return; + } + } + + assert(pWrapper == pRequest->pWrapper); + doAsyncExec(pRequest, code); +} + +void taosAsyncExecLiteral(TAOS_STMT2 *stmt) { + STscStmt2* pStmt = (STscStmt2*)stmt; + doAsyncExec(pStmt->exec.pRequest, pStmt->ctx.code); +} + void restartAsyncQuery(SRequestObj *pRequest, int32_t code) { tscInfo("restart request:%s p:%p", pRequest->sqlstr, pRequest); SRequestObj *pUserReq = pRequest; @@ -2815,6 +2838,10 @@ int taos_stmt2_bind_param(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t col int32_t code = TSDB_CODE_SUCCESS; STMT2_DLOG_E("start to bind param"); + if (stmtIsLiteral(pStmt)) { + return stmtBindLiteral2(stmt); + } + // check query bind number bool isQuery = (STMT_TYPE_QUERY == pStmt->sql.type || (pStmt->sql.type == 0 && stmt2IsSelect(stmt))); if (isQuery) { @@ -2908,6 +2935,10 @@ int taos_stmt2_bind_param_a(TAOS_STMT2 *stmt, TAOS_STMT2_BINDV *bindv, int32_t c STscStmt2 *pStmt = (STscStmt2 *)stmt; + if (stmtIsLiteral(pStmt)) { + return stmtBindLiteral2(stmt); + } + ThreadArgs *args = (ThreadArgs *)taosMemoryMalloc(sizeof(ThreadArgs)); args->stmt = stmt; args->bindv = bindv; @@ -2978,6 +3009,12 @@ int taos_stmt2_get_fields(TAOS_STMT2 *stmt, int *count, TAOS_FIELD_ALL **fields) STscStmt2 *pStmt = (STscStmt2 *)stmt; STMT2_DLOG_E("start to get fields"); + if (stmtIsLiteral(pStmt)) { + if (count) *count = 0; + if (fields) *fields = NULL; + return TSDB_CODE_SUCCESS; + } + if (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type || (pStmt->sql.type == 0 && stmt2IsInsert(stmt))) { return stmtGetStbColFields2(stmt, count, fields); diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 64dafa438260..83f759f1717a 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -13,6 +13,55 @@ char* gStmt2StatusStr[] = {"unknown", "init", "prepare", "settbname", "settags", "fetchFields", "bind", "bindCol", "addBatch", "exec"}; +#define SET_ERR(fmt, ...) do { \ + char *sbuf = pStmt->msgBuf; \ + size_t nlen = sizeof(pStmt->msgBuf); \ + if (pStmt->exec.pRequest) { \ + sbuf = pStmt->exec.pRequest->msgBuf; \ + nlen = pStmt->exec.pRequest->msgBufLen; \ + } \ + (void)snprintf(sbuf, nlen, "%s[%d]%s():" fmt "", \ + __FILE__, __LINE__, __func__, ##__VA_ARGS__); \ +} while (0) + +static inline void +stmt2LiteralCtxReset(SStmt2LiteralCtx *ctx) { + ctx->code = 0; + ctx->prepared = 0; + ctx->executing = 0; + ctx->executed = 0; + ctx->has_result_set = 0; +} + +static inline void +stmt2LiteralCtxRelease(SStmt2LiteralCtx *ctx) { + stmt2LiteralCtxReset(ctx); + if (ctx->sem_valid) { + tsem_destroy(&ctx->sem); + ctx->sem_valid = 0; + } +} + +static inline int +stmt2LiteralCtxInit(SStmt2LiteralCtx *ctx) { + if (ctx->sem_valid) return 0; + if (tsem_init(&ctx->sem, 0, 0)) return -1; + ctx->sem_valid = 1; + return 0; +} + +static inline int +stmt2LiteralCtxIsValid(SStmt2LiteralCtx *ctx) { + return ctx && ctx->sem_valid; +} + +int stmtIsLiteral(TAOS_STMT2 *stmt) { + STscStmt2* pStmt = (STscStmt2*)stmt; + return pStmt && pStmt->literal; +} + + + /* Free any existing siInfo.dbname and replace with a heap copy of src. * src may be NULL or empty — in either case dbname is left NULL. */ static int32_t stmtDupSiInfoDbname(SStbInterlaceInfo* pSi, const char* src) { @@ -2101,8 +2150,13 @@ static int32_t stmtDeepReset(STscStmt2* pStmt) { (void)taosThreadMutexDestroy(&pStmt->queue.mutex); } + // NOTE: do NOT reset until asynchronous operations have completed + stmt2LiteralCtxReset(&pStmt->ctx); + pStmt->literal = 0; + // Clean all SQL and execution info (stmtCleanSQLInfo already handles most cleanup) pStmt->bInfo.boundColsCached = false; + pStmt->bInfo.tbNameFlag = 0; // NOTE: if (stbInterlaceMode) { pStmt->bInfo.tagsCached = false; } @@ -2149,6 +2203,90 @@ static int32_t stmtDeepReset(STscStmt2* pStmt) { return TSDB_CODE_SUCCESS; } +static void stmtLiteralCallback(void *param, TAOS_RES *res, int code) { + TAOS_STMT2* stmt = (TAOS_STMT2*)param; + STscStmt2* pStmt = (STscStmt2*)stmt; + + pStmt->ctx.code = code; // NOTE: currently taos_stmt2_xxx is NOT thread-safe + + if (pStmt->exec.pRequest == NULL) { + // NOTE: preparing stage for literal statement by stmt2 + // transfer `res` which is created by buildRequest + pStmt->exec.pRequest = res; + // NOTE: wake up waiting thread + tsem_post(&pStmt->ctx.sem); + } else { + // NOTE: executing stage for literal statement by stmt2 + if (pStmt->exec.pRequest != res) { + // NOTE: internal logic error, not recoverable!!! + STMT2_ELOG("%s[%d]:%s():internal logic error", + __FILE__, __LINE__, __func__); + abort(); + } + + int nr_fields = taos_num_fields(pStmt->exec.pRequest); + if (nr_fields || + (pStmt->exec.pRequest && + pStmt->exec.pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT)) { + // NOTE: literal sql statement generates a result set + // 1. normal query with result set + // 2. empty result when `QueryTbNotExistAsEmpty` is set + // and table not exists + pStmt->ctx.has_result_set = 1; + } + + if (pStmt->options.asyncExecFn) { + // NOTE: user requires asynchronous execution via `taos_stmt2_init` + // TODO: a well-defined reentrancy protection is desired, but ... + + // NOTE: `executing` and `executed` are mutually exclusive + pStmt->ctx.executing = 0; + pStmt->ctx.executed = 1; + + pStmt->options.asyncExecFn(pStmt->options.userdata, + pStmt->exec.pRequest, pStmt->ctx.code); + } else { + // NOTE: wake up waiting thread + tsem_post(&pStmt->ctx.sem); + } + } +} + +static int stmtPrepareLiteral2(TAOS_STMT2* stmt) { + STscStmt2* pStmt = (STscStmt2*)stmt; + int32_t code = 0; + + pStmt->literal = 1; + + if (stmt2LiteralCtxInit(&pStmt->ctx)) { + SET_ERR("out of memory"); + STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY); + } + + uint64_t connId = pStmt->taos->id; + const char *sql = pStmt->sql.sqlStr; + int64_t reqid = pStmt->options.reqid; + + STscObj *pObj = acquireTscObj(connId); + if (pObj != pStmt->taos) { + releaseTscObj(connId); + SET_ERR("internal logic error"); + STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); // TODO: a new error code? + } + + taosAsyncQueryImplWithReqid(stmt, connId, sql, + stmtLiteralCallback, stmt, false, reqid); + tsem_wait(&pStmt->ctx.sem); + + releaseTscObj(connId); + + if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { + pStmt->ctx.prepared = 1; + } + + return pStmt->ctx.code; +} + int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) { STscStmt2* pStmt = (STscStmt2*)stmt; int32_t code = 0; @@ -2182,6 +2320,11 @@ int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) { STMT_ERR_RET(terrno); } pStmt->sql.sqlLen = length; + + if (qIsLiteralSql(pStmt->sql.sqlStr)) { + return stmtPrepareLiteral2(stmt); + } + STMT_ERR_RET(stmtCreateRequest(pStmt)); if (stmt2IsInsert(pStmt)) { @@ -2203,6 +2346,12 @@ int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) { } } + int count = 0; + TAOS_FIELD_ALL *fields = NULL; + code = taos_stmt2_get_fields(stmt, &count, &fields); + taos_stmt2_free_fields(stmt, fields); + fields = NULL; + STMT_ERR_RET(code); } else if (stmt2IsSelect(pStmt)) { pStmt->sql.stbInterlaceMode = false; STMT_ERR_RET(stmtParseSql(pStmt)); @@ -2212,6 +2361,13 @@ int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) { return TSDB_CODE_SUCCESS; } +int stmtBindLiteral2(TAOS_STMT2 *stmt) { + STscStmt2* pStmt = (STscStmt2*)stmt; + + SET_ERR("no data binding required for literal sql statement"); + STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR); +} + static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) { STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName)); if (!pSrc) { @@ -3349,6 +3505,62 @@ static void asyncQueryCb(void* userdata, TAOS_RES* res, int code) { } } +static int stmtExecLiteral2(TAOS_STMT2* stmt, int *affected_rows) { + STscStmt2* pStmt = (STscStmt2*)stmt; + int32_t code = 0; + + if (pStmt->ctx.code) { + return pStmt->ctx.code; + } + + if (pStmt->ctx.prepared == 0) { + SET_ERR("literal sql statement not fully prepared yet"); + STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); + } + + if (pStmt->ctx.executing) { + SET_ERR("previous execution of literal sql statement still in progress"); + STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); + } + + if (pStmt->ctx.executed) { + SET_ERR("multiple execution of a prepared literal sql statement " + "not supported yet"); + STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); + } + + pStmt->ctx.executing = 1; + + // NOTE: triggering execution logic of a prepared literal sql statement + taosAsyncExecLiteral(stmt); + + if (pStmt->options.asyncExecFn == NULL) { + // NOTE: waiting for execution process to finish + tsem_wait(&pStmt->ctx.sem); + + // NOTE: `executing` and `executed` are mutualy exclusive + pStmt->ctx.executing = 0; + pStmt->ctx.executed = 1; + + if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { + if (affected_rows) { + if (pStmt->ctx.has_result_set) { + // NOTE: literal sql statement generates a result set + *affected_rows = 0; + } else { + // NOTE: literal sql statement does not generate any result set + TAOS_RES *res = pStmt->exec.pRequest; + *affected_rows = taos_affected_rows(res); + } + } + } + return pStmt->ctx.code; + } + + // NOTE: what if taosAsyncExecLiteral failed prematurelly? + return TSDB_CODE_SUCCESS; +} + int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { STscStmt2* pStmt = (STscStmt2*)stmt; int32_t code = 0; @@ -3360,6 +3572,10 @@ int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { return pStmt->errCode; } + if (stmtIsLiteral(pStmt)) { + return stmtExecLiteral2(stmt, affected_rows); + } + STMT_ERR_RET(taosThreadMutexLock(&pStmt->asyncBindParam.mutex)); while (atomic_load_8((int8_t*)&pStmt->asyncBindParam.asyncBindNum) > 0) { (void)taosThreadCondWait(&pStmt->asyncBindParam.waitCond, &pStmt->asyncBindParam.mutex); @@ -3532,6 +3748,9 @@ int stmtClose2(TAOS_STMT2* stmt) { } } + // NOTE: do NOT release until asynchronous operations have completed + stmt2LiteralCtxRelease(&pStmt->ctx); + STMT2_DLOG("stbInterlaceMode:%d, statInfo: ctgGetTbMetaNum=>%" PRId64 ", getCacheTbInfo=>%" PRId64 ", parseSqlNum=>%" PRId64 ", pStmt->stat.bindDataNum=>%" PRId64 ", settbnameAPI:%u, bindAPI:%u, addbatchAPI:%u, execAPI:%u" @@ -3562,6 +3781,18 @@ int stmtClose2(TAOS_STMT2* stmt) { const char* stmt2Errstr(TAOS_STMT2* stmt) { STscStmt2* pStmt = (STscStmt2*)stmt; + if (stmt && stmtIsLiteral(pStmt)) { + if (NULL == pStmt->exec.pRequest) { + // NOTE: since pStmt->exec.pRequest not fully prepared yet + // error msg is stored in pStmt->msgBuf via `SET_ERR` + return pStmt->msgBuf; + } else if (pStmt->exec.pRequest->msgBuf[0]) { + // NOTE: reuse error msg stored in `msgBuf` + return pStmt->exec.pRequest->msgBuf; + } + return (const char*)tstrerror(pStmt->exec.pRequest->code); + } + if (stmt == NULL || NULL == pStmt->exec.pRequest) { return (char*)tstrerror(terrno); } @@ -3628,8 +3859,20 @@ int stmtParseColFields2(TAOS_STMT2* stmt) { } int stmtGetStbColFields2(TAOS_STMT2* stmt, int* nums, TAOS_FIELD_ALL** fields) { + STscStmt2* pStmt = (STscStmt2*)stmt; int32_t code = stmtParseColFields2(stmt); if (code != TSDB_CODE_SUCCESS) { + if (code == TSDB_CODE_TSC_STMT_TBNAME_ERROR) { + SPureInsertParserCtx ctx = {0}; + const char *pStr = pStmt->sql.sqlStr; + code = qPureParseInsert(&ctx, pStr); + if (code) { + SET_ERR("%s", ctx.buf); + return code; + } + if (nums) *nums = ctx.nr_params; + return TSDB_CODE_SUCCESS; + } return code; } @@ -3682,6 +3925,24 @@ TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) { STMT2_TLOG_E("start to use result"); + if (stmtIsLiteral(pStmt)) { + if (pStmt->ctx.executing) { + SET_ERR("literal sql statement still in progress"); + pStmt->errCode = TSDB_CODE_TSC_STMT_API_ERROR; + return NULL; + } + if (pStmt->ctx.executed == 0) { + SET_ERR("literal sql statement not executed yet"); + pStmt->errCode = TSDB_CODE_TSC_STMT_API_ERROR; + return NULL; + } + if (!pStmt->ctx.has_result_set) { + STMT2_ELOG_E("useResult only for query statement"); + return NULL; + } + return pStmt->exec.pRequest; + } + if (STMT_TYPE_QUERY != pStmt->sql.type) { STMT2_ELOG_E("useResult only for query statement"); return NULL; @@ -3753,4 +4014,4 @@ int32_t stmtBuildErrorMsgWithCode(STscStmt2* pStmt, const char* msg, int32_t err pStmt->errCode = errorCode; return errorCode; -} \ No newline at end of file +} diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index 944e4efc9e5a..dbaba68787a1 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -124,7 +124,10 @@ void getFieldsError(TAOS* taos, const char* sql, int errorCode, const char* erro TAOS_STMT2* stmt = taos_stmt2_init(taos, &option); ASSERT_NE(stmt, nullptr); int code = taos_stmt2_prepare(stmt, sql, 0); - checkError(stmt, code, __FILE__, __LINE__); + // checkError(stmt, code, __FILE__, __LINE__); + ASSERT_EQ(code, errorCode); + + if (code) return; int fieldNum = 0; TAOS_FIELD_ALL* pFields = NULL; @@ -477,6 +480,8 @@ TEST(stmt2Case, timezone) { getRecordCounts++; } ASSERT_EQ(getRecordCounts, 1); + + taos_stmt2_close(stmt); } // stmt2 wiht time str in UTC timezone @@ -781,7 +786,7 @@ TEST(stmt2Case, insert_stb_get_fields_Test) { { const char* sql = "insert into stmt2_testdb_2.stb(t1,t2,ts,b,tbname) values(*,*,*,*,*)"; printf("case 9 : %s\n", sql); - getFieldsError(taos, sql, TSDB_CODE_PAR_INVALID_COLUMNS_NUM, "Illegal number of columns"); + getFieldsError(taos, sql, /*TSDB_CODE_PAR_INVALID_COLUMNS_NUM*/TSDB_CODE_TSC_SQL_SYNTAX_ERROR, "Illegal number of columns"); } do_query(taos, "drop database if exists stmt2_testdb_2"); @@ -1120,7 +1125,7 @@ TEST(stmt2Case, insert_ntb_get_fields_Test) { { const char* sql = "insert into stmt2_testdb_4.? values(?,?)"; printf("case 2 : %s\n", sql); - getFieldsError(taos, sql, TSDB_CODE_TSC_STMT_TBNAME_ERROR, "Table does not exist"); + getFieldsError(taos, sql, /*TSDB_CODE_TSC_STMT_TBNAME_ERROR*/TSDB_CODE_SUCCESS, "Table does not exist"); } // case 3 : wrong para nums @@ -1610,6 +1615,7 @@ TEST(stmt2Case, stmt2_insert_non_statndard) { ASSERT_EQ(code, TSDB_CODE_PAR_SYNTAX_ERROR); ASSERT_STREQ(taos_stmt2_error(stmt), "stmt only support 'SELECT' or 'INSERT'"); + taos_stmt2_close(stmt); } @@ -3747,7 +3753,7 @@ TEST(stmt2Case, errcode) { int fieldNum = 0; TAOS_FIELD_ALL* pFields = NULL; code = taos_stmt2_get_fields(stmt, &fieldNum, &pFields); - ASSERT_EQ(code, TSDB_CODE_TSC_STMT_TBNAME_ERROR); + ASSERT_EQ(code, /*TSDB_CODE_TSC_STMT_TBNAME_ERROR*/TSDB_CODE_SUCCESS); // get fail dont influence the next stmt prepare sql = "insert into ? (ts, name) values (?, ?)"; @@ -5124,7 +5130,7 @@ TEST(stmt2Case, no_tag) { TAOS_STMT2_BIND* paramv = ¶ms[0]; TAOS_STMT2_BINDV bindv = {1, &tbname[1], NULL, ¶mv}; code = taos_stmt2_bind_param(stmt, &bindv, -1); - ASSERT_EQ(code, TSDB_CODE_PAR_TABLE_NOT_EXIST); + ASSERT_EQ(code, /*TSDB_CODE_PAR_TABLE_NOT_EXIST*/TSDB_CODE_TSC_STMT_CACHE_ERROR); taos_stmt2_close(stmt); } @@ -5145,7 +5151,7 @@ TEST(stmt2Case, no_tag) { TAOS_STMT2_BIND* paramv = ¶ms[0]; TAOS_STMT2_BINDV bindv = {1, &tbname[1], NULL, ¶mv}; code = taos_stmt2_bind_param(stmt, &bindv, -1); - ASSERT_EQ(code, TSDB_CODE_PAR_TABLE_NOT_EXIST); + ASSERT_EQ(code, /*TSDB_CODE_PAR_TABLE_NOT_EXIST*/TSDB_CODE_TSC_STMT_CACHE_ERROR); taos_stmt2_close(stmt); } @@ -5208,7 +5214,7 @@ TEST(stmt2Case, no_tag) { TAOS_STMT2_BIND* paramv[2] = {¶ms[0], ¶ms[0]}; TAOS_STMT2_BINDV bindv = {1, &tbname[1], &pTag, ¶mv[0]}; code = taos_stmt2_bind_param(stmt, &bindv, -1); - ASSERT_EQ(code, TSDB_CODE_PAR_TABLE_NOT_EXIST); + ASSERT_EQ(code, /*TSDB_CODE_PAR_TABLE_NOT_EXIST*/TSDB_CODE_TSC_STMT_CACHE_ERROR); taos_stmt2_close(stmt); } @@ -5226,7 +5232,7 @@ TEST(stmt2Case, no_tag) { TAOS_STMT2_BIND* paramv[2] = {¶ms[0], ¶ms[0]}; TAOS_STMT2_BINDV bindv = {1, &tbname[1], &pTag, ¶mv[0]}; code = taos_stmt2_bind_param(stmt, &bindv, -1); - ASSERT_EQ(code, TSDB_CODE_PAR_TABLE_NOT_EXIST); + ASSERT_EQ(code, /*TSDB_CODE_PAR_TABLE_NOT_EXIST*/TSDB_CODE_TSC_STMT_CACHE_ERROR); taos_stmt2_close(stmt); } @@ -6163,4 +6169,289 @@ TEST(stmt2Case, query_timestamp_auto_precision) { taos_close(taos); } +class stmt2CaseF : public testing::Test { + public: + stmt2CaseF() : taos_(NULL), stmt2_(NULL) { } + ~stmt2CaseF() { } + + void SetUp(void) override { + } + + void TearDown(void) override { + if (stmt2_) { + taos_stmt2_close(stmt2_); + stmt2_ = NULL; + } + if (taos_) { + taos_close(taos_); + taos_ = NULL; + } + } + + const char* hexify(int32_t code) { + snprintf(buf_, sizeof(buf_), "0x%08x", code); + return buf_; + } + + TAOS *taos_; + TAOS_STMT2 *stmt2_; + char buf_[64]; +}; + +TEST_F(stmt2CaseF, exec_direct) { + taos_ = taos_connect("localhost", "root", "taosdata", "", 0); + ASSERT_NE(taos_, nullptr); + + stmt2_ = taos_stmt2_init(taos_, NULL); + ASSERT_NE(stmt2_, nullptr); + +#define R(sql, exp, affected_rows, rows) { __LINE__, sql, exp, affected_rows, rows } + struct { + int line; + const char *sql; + bool exp_ok; + int exp_affected_rows; + int exp_rows; + } _cases[] = { + R("drop database if exists stmt2_exec_direct", + true, 0, -1), + R("create database if not exists stmt2_exec_direct precision 'ns'", + true, 0, -1), + R("create table stmt2_exec_direct.t (ts timestamp, i32 int)", + true, 0, -1), + R("insert into stmt2_exec_direct.t (ts, i32) values (now, 1) (now+1b, 2) (now+2b, 3)", + true, 3, -1), + R("select * from stmt2_exec_direct.t", + true, 0, 3), + R("select * from stmt2_exec_direct.t where 1 = 2", + true, 0, 0), + }; +#undef R + for (size_t i=0; ibuf, sizeof(pCtx->buf), \ + "expecting %s, but got `%.*s`", \ + msg, token.n < 10 ? 10 : token.n, token.z); \ + return TSDB_CODE_PAR_SYNTAX_ERROR; \ +} while (0) + +static inline const char* qNextToken(const char *pStr, SToken *t) { +again: + t->z = (char*)pStr; + t->n = tGetToken(pStr, &t->type, NULL); + if (t->n == 0) return pStr; + if (t->type != TK_NK_SPACE) return pStr + t->n; + pStr += t->n; + goto again; +} + +int32_t qPureParseInsert(SPureInsertParserCtx *pCtx, const char *pStr) { + // NOTE: only for insert into [db.]? (...) values (...) + pCtx->nr_params = 0; + + int nr_names = 0; + int nr_values = 0; + int32_t code = TSDB_CODE_SUCCESS; + + const char* pSql = pStr; + SToken token; + + ADVANCE(); + if (!MATCH(TK_INSERT)) { + RETURN_EXPECTING("`insert`"); + } + + ADVANCE(); + if (!MATCH(TK_INTO)) { + RETURN_EXPECTING("`into`"); + } + + ADVANCE(); + if (MATCH(TK_NK_ID)) { + SToken db = token; + ADVANCE(); + if (!MATCH(TK_NK_DOT)) { + RETURN_EXPECTING("`.`"); + } + ADVANCE(); + if (!MATCH(TK_NK_QUESTION)) { + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; + } + if (db.z + db.n + 1 + token.n != token.z + token.n) { + token.n = token.z + token.n - db.z; + token.z = db.z; + RETURN_EXPECTING(".?"); + } + ++pCtx->nr_params; + } else if (MATCH(TK_NK_QUESTION)) { + ++pCtx->nr_params; + } else { + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; + } + + ADVANCE(); + if (MATCH(TK_USING)) { + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; + } + if (MATCH(TK_NK_LP)) { + ADVANCE(); +again: + if (!MATCH(TK_NK_ID)) { + RETURN_EXPECTING(""); + } + ++nr_names; + ADVANCE(); + if (MATCH(TK_NK_COMMA)) { + ADVANCE(); + if (!MATCH(TK_NK_ID)) { + RETURN_EXPECTING("``"); + } + goto again; + } else if (!MATCH(TK_NK_RP)) { + RETURN_EXPECTING("`,` or `)`"); + } + ADVANCE(); + } else if (!MATCH(TK_VALUES)) { + RETURN_EXPECTING("`(` or `values`"); + } + + if (!MATCH(TK_VALUES)) { + if (MATCH(TK_USING)) { + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; + } + RETURN_EXPECTING("`values`"); + } + + ADVANCE(); + if (!MATCH(TK_NK_LP)) { + RETURN_EXPECTING("`(`"); + } + + ADVANCE(); + +next_value: + if (MATCH(TK_NK_QUESTION)) { + ++pCtx->nr_params; + ++nr_values; + ADVANCE(); + } else if (MATCH(TK_NK_INTEGER) + || MATCH(TK_NK_FLOAT) + || MATCH(TK_NK_STRING) + || MATCH(TK_NK_BOOL) + || MATCH(TK_NULL)) { + if (nr_values == nr_names) { + RETURN_EXPECTING("`)`"); + } + ++nr_values; + ADVANCE(); + } else if (MATCH(TK_NK_MINUS)) { + SToken t1 = token; + ADVANCE(); + if (!MATCH(TK_NK_INTEGER) && !MATCH(TK_NK_FLOAT)) { + RETURN_EXPECTING("|"); + } + if (t1.n + token.n != token.z + token.n - t1.z) { + token = t1; + RETURN_EXPECTING("|"); + } + ADVANCE(); + ++nr_values; + } else if (MATCH(TK_NOW)) { + ADVANCE(); + if (MATCH(TK_NK_LP)) { + ADVANCE(); + if (!MATCH(TK_NK_RP)) { + RETURN_EXPECTING("`)`"); + } + ADVANCE(); + } + ++nr_values; + } else if (MATCH(TK_TODAY)) { + ADVANCE(); + if (MATCH(TK_NK_LP)) { + ADVANCE(); + if (!MATCH(TK_NK_RP)) { + RETURN_EXPECTING("`)`"); + } + ADVANCE(); + } + ++nr_values; + } else { + RETURN_EXPECTING("`?` or or or "); + } + + if (MATCH(TK_NK_COMMA)) { + if (nr_values == nr_names) { + RETURN_EXPECTING("`)`"); + } + ADVANCE(); + goto next_value; + } else if (MATCH(TK_NK_RP)) { + if (nr_names && nr_values != nr_names) { + RETURN_EXPECTING("`,`"); + } + } else { + if (nr_values == nr_names) { + RETURN_EXPECTING("`)`"); + } else { + RETURN_EXPECTING("`,`"); + } + } + + ADVANCE(); + if (MATCH(TK_NK_SEMI)) return TSDB_CODE_SUCCESS; + if (token.n) { + RETURN_EXPECTING(""); + } + + return TSDB_CODE_SUCCESS; +} + bool qIsInsertValuesSql(const char* pStr, size_t length) { if (NULL == pStr) { return false; diff --git a/source/util/src/thash.c b/source/util/src/thash.c index 236c01f33f40..8f6619b20372 100644 --- a/source/util/src/thash.c +++ b/source/util/src/thash.c @@ -677,7 +677,7 @@ SHashNode *doCreateHashNode(const void *key, size_t keyLen, const void *pData, s #ifdef _TD_LOONGARCH_64 SHashNode *pNewNode = taosMemoryCalloc(1, sizeof(SHashNode) + keyLen + dsize + 1); #else - SHashNode *pNewNode = taosMemoryMalloc(sizeof(SHashNode) + keyLen + dsize + 1); + SHashNode *pNewNode = taosMemoryCalloc(1, sizeof(SHashNode) + keyLen + dsize + 1); #endif if (pNewNode == NULL) {