From 332f4320633f702ba4de83766232854064e626b3 Mon Sep 17 00:00:00 2001 From: freemine Date: Mon, 25 May 2026 22:04:52 +0800 Subject: [PATCH 01/16] feat(stmt2): literal sql statement can be exec-direct, that is taos_stmt2_prepare followed by taos_stmt2_exec --- include/libs/parser/parser.h | 1 + source/client/inc/clientInt.h | 19 +- source/client/inc/clientStmt2.h | 18 ++ source/client/src/clientImpl.c | 43 +++-- source/client/src/clientMain.c | 71 ++++++-- source/client/src/clientStmt2.c | 172 +++++++++++++++++- source/client/test/stmt2Test.cpp | 127 +++++++++++++ .../dnode/mnode/impl/test/sma/CMakeLists.txt | 2 +- .../dnode/mnode/impl/test/stb/CMakeLists.txt | 2 +- source/libs/parser/src/parser.c | 19 ++ 10 files changed, 434 insertions(+), 40 deletions(-) diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 84f1da167174..9f863a2ab871 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -166,6 +166,7 @@ typedef struct SParseContext { } SParseContext; int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery); +bool qIsLiteralSql(const char* pStr, size_t length); 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..dec965f7fbb0 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -37,6 +37,19 @@ extern "C" { // #include "clientSession.h" #include "tconfig.h" +#if 0 /* { */ +#define D(fmt, ...) \ + fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ + (void*)(uintptr_t)pthread_self(), __FILE__, __LINE__, __func__, \ + ##__VA_ARGS__) + +#define A(expr, fmt, ...) \ + if (!(expr)) { \ + D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ + abort(); \ + } +#endif /* } */ + #define ERROR_MSG_BUF_DEFAULT_SIZE 512 #define HEARTBEAT_INTERVAL 1500 // ms @@ -364,6 +377,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 +402,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..cae6376ab56c 100644 --- a/source/client/inc/clientStmt2.h +++ b/source/client/inc/clientStmt2.h @@ -86,6 +86,16 @@ STableDataCxt *pCurrBlock; SSubmitTbData *pCurrTbData; } SStmtExecInfo; */ +typedef struct SStmt2LiteralCtx { + tsem_t sem; + + int32_t code; + + uint8_t sem_valid:1; + uint8_t prepared:1; + uint8_t executed:1; +} SStmt2LiteralCtx; + typedef struct { bool stbInterlaceMode; STMT_TYPE type; @@ -176,6 +186,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 +270,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 +286,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..674ca149edf8 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,29 +3462,36 @@ void doRequestCallback(SRequestObj* pRequest, int32_t code) { pRequest->inCallback = true; int64_t this = pRequest->self; - 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; - pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT; - if (pRequest->code == TSDB_CODE_PAR_TABLE_NOT_EXIST || pRequest->code == TSDB_CODE_TDB_TABLE_NOT_EXIST) { - pRequest->code = TSDB_CODE_SUCCESS; - if (pRequest->msgBuf != NULL && pRequest->msgBufLen > 0) { - pRequest->msgBuf[0] = '\0'; + + if (!pRequest->literal_by_stmt2) { + 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; + pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT; + if (pRequest->code == TSDB_CODE_PAR_TABLE_NOT_EXIST || pRequest->code == TSDB_CODE_TDB_TABLE_NOT_EXIST) { + pRequest->code = TSDB_CODE_SUCCESS; + if (pRequest->msgBuf != NULL && pRequest->msgBufLen > 0) { + pRequest->msgBuf[0] = '\0'; + } } } - } - tscDebug("QID:0x%" PRIx64 ", taos_query end, req:0x%" PRIx64 ", res:%p", pRequest->requestId, pRequest->self, - pRequest); + tscDebug("QID:0x%" PRIx64 ", taos_query end, req:0x%" PRIx64 ", res:%p", pRequest->requestId, pRequest->self, + pRequest); + } if (pRequest->body.queryFp != NULL) { pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code); } - SRequestObj* pReq = acquireRequest(this); - if (pReq != NULL) { - pReq->inCallback = false; - (void)releaseRequest(this); + if (!pRequest->literal_by_stmt2) { + SRequestObj* pReq = acquireRequest(this); + if (pReq != NULL) { + pReq->inCallback = false; + (void)releaseRequest(this); + } + } else { + pRequest->inCallback = false; } } diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index b6059a96837a..7a08a7d5a122 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -2005,7 +2005,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 +2089,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 +2130,41 @@ 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 (TSDB_CODE_SUCCESS == code) { + if (pRequest->literal_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 +2834,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 +2931,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 +3005,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..9c6433ecc9be 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -13,6 +13,53 @@ 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->executed = 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) { @@ -2073,6 +2120,9 @@ static int32_t stmtDeepReset(STscStmt2* pStmt) { pStmt->errCode = 0; + stmt2LiteralCtxReset(&pStmt->ctx); + pStmt->literal = 0; + // Wait for async execution to complete if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) { if (tsem_wait(&pStmt->asyncExecSem) != 0) { @@ -2149,6 +2199,57 @@ 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; + if (pStmt->exec.pRequest == NULL) { + pStmt->exec.pRequest = res; + } else { + // NOTE: internal logic error, not recoverable!!! + if (pStmt->exec.pRequest != res) { + STMT2_ELOG("%s[%d]:%s():internal logic error", + __FILE__, __LINE__, __func__); + abort(); + } + } + pStmt->ctx.code = code; + 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) { + 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 +2283,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, pStmt->sql.sqlLen)) { + return stmtPrepareLiteral2(stmt); + } + STMT_ERR_RET(stmtCreateRequest(pStmt)); if (stmt2IsInsert(pStmt)) { @@ -2212,6 +2318,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_ERR_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 +3462,41 @@ 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.executed) { + SET_ERR("multiple execution of literal sql statement not supported yet"); + STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); + } + + pStmt->ctx.executed = 1; + + taosAsyncExecLiteral(stmt); + tsem_wait(&pStmt->ctx.sem); + + if (affected_rows) { + if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { + TAOS_RES *res = pStmt->exec.pRequest; + *affected_rows = taos_affected_rows(res); + } else { + *affected_rows = 0; + } + } + + return pStmt->ctx.code; +} + int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { STscStmt2* pStmt = (STscStmt2*)stmt; int32_t code = 0; @@ -3360,6 +3508,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); @@ -3518,6 +3670,8 @@ int stmtClose2(TAOS_STMT2* stmt) { (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond); (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex); + stmt2LiteralCtxRelease(&pStmt->ctx); + if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) { if (tsem_wait(&pStmt->asyncExecSem) != 0) { STMT2_ELOG_E("fail to wait asyncExecSem"); @@ -3563,6 +3717,9 @@ const char* stmt2Errstr(TAOS_STMT2* stmt) { STscStmt2* pStmt = (STscStmt2*)stmt; if (stmt == NULL || NULL == pStmt->exec.pRequest) { + if (stmt && stmtIsLiteral(pStmt)) { + return pStmt->msgBuf; + } return (char*)tstrerror(terrno); } @@ -3682,6 +3839,19 @@ TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) { STMT2_TLOG_E("start to use result"); + if (stmtIsLiteral(pStmt)) { + if (pStmt->ctx.executed == 0) { + SET_ERR("literal sql statement not executed yet"); + pStmt->errCode = TSDB_CODE_TSC_STMT_API_ERROR; + return NULL; + } + if (taos_num_fields(pStmt->exec.pRequest) == 0) { + STMT2_ELOG_E("useResult only for query statement even it's literal"); + 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 +3923,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..4831cb6181e5 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -6163,4 +6163,131 @@ 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; + } + } + + TAOS *taos_; + TAOS_STMT2 *stmt2_; +}; + +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 foo", true, 0, -1), + R("create database if not exists foo precision 'ns'", true, 0, -1), + R("create table foo.t (ts timestamp, i32 int)", true, 0, -1), + R("insert into foo.t (ts, i32) values (now, 1) (now+1b, 2) (now+2b, 3)", true, 3, -1), + R("select * from foo.t", true, 0, 3), + R("select * from foo.t where 1 = 2", true, 0, 0), + }; +#undef R + for (size_t i=0; i Date: Tue, 26 May 2026 00:18:27 +0800 Subject: [PATCH 02/16] feat(stmt2): fully prepared for most insert statement, except for insert into ? (...) values (...) --- source/client/src/clientStmt2.c | 6 ++ source/client/test/stmt2Test.cpp | 137 ++++++++++++++++++++++++++++--- 2 files changed, 133 insertions(+), 10 deletions(-) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 9c6433ecc9be..4b70423a0d5e 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2309,6 +2309,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)); diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index 4831cb6181e5..a36307d08c8a 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -6182,8 +6182,14 @@ class stmt2CaseF : public testing::Test { } } + 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) { @@ -6201,12 +6207,18 @@ TEST_F(stmt2CaseF, exec_direct) { int exp_affected_rows; int exp_rows; } _cases[] = { - R("drop database if exists foo", true, 0, -1), - R("create database if not exists foo precision 'ns'", true, 0, -1), - R("create table foo.t (ts timestamp, i32 int)", true, 0, -1), - R("insert into foo.t (ts, i32) values (now, 1) (now+1b, 2) (now+2b, 3)", true, 3, -1), - R("select * from foo.t", true, 0, 3), - R("select * from foo.t where 1 = 2", true, 0, 0), + 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; i Date: Tue, 26 May 2026 12:02:03 +0800 Subject: [PATCH 03/16] feat(stmt2): literal sql statement can be exec-direct in an async mode --- source/client/inc/clientInt.h | 18 +++++++++++++++++- source/client/src/clientStmt2.c | 6 ++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index dec965f7fbb0..752e67fd6fde 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -37,10 +37,25 @@ extern "C" { // #include "clientSession.h" #include "tconfig.h" +#include + + +#if 0 /* { */ #if 0 /* { */ #define D(fmt, ...) \ fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ - (void*)(uintptr_t)pthread_self(), __FILE__, __LINE__, __func__, \ + (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ + ##__VA_ARGS__) + +#define A(expr, fmt, ...) \ + if (!(expr)) { \ + D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ + abort(); \ + } +#else /* }{ */ +#define D(fmt, ...) \ + syslog(LOG_DEBUG, "@[%p]%s[%d]:%s():" fmt "\n", \ + (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ ##__VA_ARGS__) #define A(expr, fmt, ...) \ @@ -49,6 +64,7 @@ extern "C" { abort(); \ } #endif /* } */ +#endif /* } */ #define ERROR_MSG_BUF_DEFAULT_SIZE 512 #define HEARTBEAT_INTERVAL 1500 // ms diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 4b70423a0d5e..ad8926cd5db4 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -3500,6 +3500,12 @@ static int stmtExecLiteral2(TAOS_STMT2* stmt, int *affected_rows) { } } + if (pStmt->options.asyncExecFn) { + // TODO: a well-defined reentrancy protection is desired, but ... + pStmt->options.asyncExecFn(pStmt->options.userdata, + pStmt->exec.pRequest, pStmt->ctx.code); + } + return pStmt->ctx.code; } From dc9afb56fa1a977581dacbe9818eefcd84708a3b Mon Sep 17 00:00:00 2001 From: freemine Date: Wed, 27 May 2026 00:05:21 +0800 Subject: [PATCH 04/16] feat(stmt2): insert into [db.]? (...) values (...) also fully prepared during `taos_stmt2_prepare` --- include/libs/parser/parser.h | 38 ++++++++ source/client/inc/clientInt.h | 29 ------ source/client/src/clientStmt2.c | 12 +++ source/client/test/stmt2Test.cpp | 37 +++++--- source/libs/parser/src/parser.c | 146 +++++++++++++++++++++++++++++++ 5 files changed, 219 insertions(+), 43 deletions(-) diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 9f863a2ab871..7a40009b3577 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -24,6 +24,34 @@ extern "C" { #include "query.h" #include "querynodes.h" +#include + +#if 0 /* { */ +#if 0 /* { */ +#define D(fmt, ...) \ + fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ + (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ + ##__VA_ARGS__) + +#define A(expr, fmt, ...) \ + if (!(expr)) { \ + D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ + abort(); \ + } +#else /* }{ */ +#define D(fmt, ...) \ + syslog(LOG_DEBUG, "@[%p]%s[%d]:%s():" fmt "\n", \ + (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ + ##__VA_ARGS__) + +#define A(expr, fmt, ...) \ + if (!(expr)) { \ + D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ + abort(); \ + } +#endif /* } */ +#endif /* } */ + #define PAR_ERR_RET(c) \ do { \ int32_t _code = c; \ @@ -165,8 +193,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, size_t length); + +// 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 752e67fd6fde..8cfab62a7928 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -37,35 +37,6 @@ extern "C" { // #include "clientSession.h" #include "tconfig.h" -#include - - -#if 0 /* { */ -#if 0 /* { */ -#define D(fmt, ...) \ - fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ - (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ - ##__VA_ARGS__) - -#define A(expr, fmt, ...) \ - if (!(expr)) { \ - D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ - abort(); \ - } -#else /* }{ */ -#define D(fmt, ...) \ - syslog(LOG_DEBUG, "@[%p]%s[%d]:%s():" fmt "\n", \ - (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ - ##__VA_ARGS__) - -#define A(expr, fmt, ...) \ - if (!(expr)) { \ - D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ - abort(); \ - } -#endif /* } */ -#endif /* } */ - #define ERROR_MSG_BUF_DEFAULT_SIZE 512 #define HEARTBEAT_INTERVAL 1500 // ms diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index ad8926cd5db4..8dfc538cc3ca 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -3797,8 +3797,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_PAR_TABLE_NOT_EXIST || 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; } diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index a36307d08c8a..0f69c0d7760b 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -6340,46 +6340,55 @@ TEST_F(stmt2CaseF, insert) { << sql << std::endl; } -#define R(sql, code, exp_params) { __LINE__, sql, code, exp_params} +#define R(sql, exp_ok, exp_params) { __LINE__, sql, exp_ok, exp_params} struct { int line; const char *sql; - int exp_code; + int exp_ok; int exp_params; } _cases1[] = { + R("insert into stmt2_insert.tx (ts, i32) values (?, ?)", + false, -1), + R("insert into stmt2_insert.xstb (tbname, tname, ts, i32) values (?, ?, ?, ?)", + false, -1), + R("insert into ? using stmt2_insert.xstb (tname) tags (?) (ts, i32) values (?, ?)", + false, -1), + R("insert into ? (ts, i32) using stmt2_insert.xstb (tname) tags (?) values (?, ?)", + false, -1), R("insert into stmt2_insert.t (ts, i32) values (?, ?)", - TSDB_CODE_SUCCESS, 2), + true, 2), R("insert into stmt2_insert.stb (tbname, tname, ts, i32) values (?, ?, ?, ?)", - TSDB_CODE_SUCCESS, 4), + true, 4), R("insert into ? using stmt2_insert.stb (tname) tags (?) (ts, i32) values (?, ?)", - TSDB_CODE_SUCCESS, 4), + true, 4), R("insert into ? (ts, i32) using stmt2_insert.stb (tname) tags (?) values (?, ?)", - TSDB_CODE_SUCCESS, 4), - R("insert into ? (tbname, tname, ts, i32) values (?, ?, ?, ?)", - TSDB_CODE_PAR_TABLE_NOT_EXIST, -1), + true, 4), + R("insert into ? (ts, i32) values (?, ?)", + true, 3), R("select * from stmt2_insert.t where ts = ? and (i32 = ? or i32 = ?)", - TSDB_CODE_SUCCESS, 3), + true, 3), R("insert into stmt2_insert.t (ts, i32) values (?, ?) (?, ?)", - TSDB_CODE_SUCCESS, 2), // flaw: redundant (?, ?) shall be reported as error + true, 2), // flaw: redundant (?, ?) shall be reported as error R("insert into stmt2_insert.t (ts, i32) values (?, 1)", - TSDB_CODE_SUCCESS, 1), + true, 1), }; #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_PAR_TABLE_NOT_EXIST; + } + 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_PAR_TABLE_NOT_EXIST; + } + + ADVANCE(); + if (MATCH(TK_USING)) { + return TSDB_CODE_PAR_TABLE_NOT_EXIST; + } + 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_PAR_TABLE_NOT_EXIST; + } + RETURN_EXPECTING("`values`"); + } + + ADVANCE(); + if (!MATCH(TK_NK_LP)) { + RETURN_EXPECTING("`(`"); + } + + ADVANCE(); + +next_value: + if (MATCH(TK_NK_QUESTION)) { + ++pCtx->nr_params; + ++nr_values; + } else if (MATCH(TK_NK_INTEGER) + || MATCH(TK_NK_FLOAT) + || MATCH(TK_NK_STRING)) { + if (nr_values == nr_names) { + RETURN_EXPECTING("`)`"); + } + ++nr_values; + } else { + RETURN_EXPECTING("`?` or or or "); + } + + ADVANCE(); + if (MATCH(TK_NK_COMMA)) { + if (nr_values == nr_names) { + RETURN_EXPECTING("`)`"); + } + ADVANCE(); + goto next_value; + } else if (MATCH(TK_NK_RP)) { + if (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; From 9729efd8a5987f3804bbe5be085abdc680c6b6ac Mon Sep 17 00:00:00 2001 From: freemine Date: Wed, 27 May 2026 17:49:49 +0800 Subject: [PATCH 05/16] fix(stmt2): several tiny bugs/typos found in parameterized-literal-prepare procedure --- include/libs/parser/parser.h | 2 +- source/client/src/clientMain.c | 8 ++--- source/client/src/clientStmt2.c | 3 +- source/client/test/stmt2Test.cpp | 50 +++++++++++++++++++++----------- source/libs/parser/src/parser.c | 8 ++--- 5 files changed, 43 insertions(+), 28 deletions(-) diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 7a40009b3577..f8b8fa57eac4 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -27,7 +27,7 @@ extern "C" { #include #if 0 /* { */ -#if 0 /* { */ +#if 1 /* { */ #define D(fmt, ...) \ fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 7a08a7d5a122..9da9c04e6307 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -2149,11 +2149,9 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { code = prepareAndParseSqlSyntax(&pWrapper, pRequest, updateMetaForce); } - if (TSDB_CODE_SUCCESS == code) { - if (pRequest->literal_by_stmt2) { - doRequestCallback(pRequest, code); - return; - } + if (pRequest->literal_by_stmt2) { + doRequestCallback(pRequest, code); + return; } assert(pWrapper == pRequest->pWrapper); diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 8dfc538cc3ca..165d2e7e202d 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2153,6 +2153,7 @@ static int32_t stmtDeepReset(STscStmt2* pStmt) { // 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; } @@ -3800,7 +3801,7 @@ 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_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_TSC_STMT_TBNAME_ERROR) { + if (code == TSDB_CODE_TSC_STMT_TBNAME_ERROR) { SPureInsertParserCtx ctx = {0}; const char *pStr = pStmt->sql.sqlStr; code = qPureParseInsert(&ctx, pStr); diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index 0f69c0d7760b..58bd459f7e3c 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -6340,55 +6340,71 @@ TEST_F(stmt2CaseF, insert) { << sql << std::endl; } -#define R(sql, exp_ok, exp_params) { __LINE__, sql, exp_ok, exp_params} +#define R(sql, exp_code, exp_params) { __LINE__, sql, exp_code, exp_params} struct { int line; const char *sql; - int exp_ok; + int32_t exp_code; int exp_params; } _cases1[] = { + R("insert into stmt2_insert.t (ts, i32) values (?, 1)", + TSDB_CODE_SUCCESS, 1), + R("insert into stmt2_insert.? (ts, i32) values (?, ?)", + TSDB_CODE_SUCCESS, 3), + R("insert into stmt2_insert.t (ts, i32) values (?, ?)", + TSDB_CODE_SUCCESS, 2), + // no table found R("insert into stmt2_insert.tx (ts, i32) values (?, ?)", - false, -1), + TSDB_CODE_TSC_STMT_TBNAME_ERROR, -1), + // no super table (guessed by `tbname`) found R("insert into stmt2_insert.xstb (tbname, tname, ts, i32) values (?, ?, ?, ?)", - false, -1), + TSDB_CODE_TSC_STMT_TBNAME_ERROR, -1), + // no super table (guessed by using clause) found R("insert into ? using stmt2_insert.xstb (tname) tags (?) (ts, i32) values (?, ?)", - false, -1), + TSDB_CODE_TSC_STMT_TBNAME_ERROR, -1), + // no super table (guessed by using clause with normal column names ahead) found R("insert into ? (ts, i32) using stmt2_insert.xstb (tname) tags (?) values (?, ?)", - false, -1), + TSDB_CODE_TSC_STMT_TBNAME_ERROR, -1), + // invalid column name + R("insert into stmt2_insert.stb(t1,t2,ts,b,tbname) values(*,*,*,*,*)", + TSDB_CODE_PAR_INVALID_COLUMN, -1), + // invalid `*` in values clause + R("insert into stmt2_insert.stb(tname,ts,i32,tbname) values(*,*,*,*)", + TSDB_CODE_TSC_SQL_SYNTAX_ERROR, -1), R("insert into stmt2_insert.t (ts, i32) values (?, ?)", - true, 2), + TSDB_CODE_SUCCESS, 2), R("insert into stmt2_insert.stb (tbname, tname, ts, i32) values (?, ?, ?, ?)", - true, 4), + TSDB_CODE_SUCCESS, 4), R("insert into ? using stmt2_insert.stb (tname) tags (?) (ts, i32) values (?, ?)", - true, 4), + TSDB_CODE_SUCCESS, 4), R("insert into ? (ts, i32) using stmt2_insert.stb (tname) tags (?) values (?, ?)", - true, 4), + TSDB_CODE_SUCCESS, 4), R("insert into ? (ts, i32) values (?, ?)", - true, 3), + TSDB_CODE_SUCCESS, 3), R("select * from stmt2_insert.t where ts = ? and (i32 = ? or i32 = ?)", - true, 3), + TSDB_CODE_SUCCESS, 3), R("insert into stmt2_insert.t (ts, i32) values (?, ?) (?, ?)", - true, 2), // flaw: redundant (?, ?) shall be reported as error + TSDB_CODE_SUCCESS, 2), // flaw: redundant (?, ?) shall be reported as error R("insert into stmt2_insert.t (ts, i32) values (?, 1)", - true, 1), + TSDB_CODE_SUCCESS, 1), }; #undef R for (size_t i=0; inr_params; } else { - return TSDB_CODE_PAR_TABLE_NOT_EXIST; + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; } ADVANCE(); if (MATCH(TK_USING)) { - return TSDB_CODE_PAR_TABLE_NOT_EXIST; + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; } if (MATCH(TK_NK_LP)) { ADVANCE(); @@ -136,7 +136,7 @@ int32_t qPureParseInsert(SPureInsertParserCtx *pCtx, const char *pStr) { if (!MATCH(TK_VALUES)) { if (MATCH(TK_USING)) { - return TSDB_CODE_PAR_TABLE_NOT_EXIST; + return TSDB_CODE_TSC_STMT_TBNAME_ERROR; } RETURN_EXPECTING("`values`"); } From 3004f938f8ab5ec84e44bb5d84ecca7590a56a4c Mon Sep 17 00:00:00 2001 From: freemine Date: Wed, 27 May 2026 22:03:47 +0800 Subject: [PATCH 06/16] fix(stmt2): adjust stmt2Test test cases accordingly, and some tiny bugfix/typos --- include/libs/parser/parser.h | 4 ++-- source/client/src/clientStmt2.c | 2 +- source/client/test/stmt2Test.cpp | 24 ++++++++++++++++-------- source/libs/parser/src/parser.c | 6 +++--- source/util/src/thash.c | 2 +- 5 files changed, 23 insertions(+), 15 deletions(-) diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index f8b8fa57eac4..428a448c40e5 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -27,7 +27,7 @@ extern "C" { #include #if 0 /* { */ -#if 1 /* { */ +#if 0 /* { */ #define D(fmt, ...) \ fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ @@ -200,7 +200,7 @@ typedef struct SPureInsertParserCtx { } SPureInsertParserCtx; int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery); -bool qIsLiteralSql(const char* pStr, size_t length); +bool qIsLiteralSql(const char* pStr); // NOTE: only for insert into [db.]? (...) values (...) int32_t qPureParseInsert(SPureInsertParserCtx *pCtx, const char *pStr); diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 165d2e7e202d..7d1129148c83 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2285,7 +2285,7 @@ int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) { } pStmt->sql.sqlLen = length; - if (qIsLiteralSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen)) { + if (qIsLiteralSql(pStmt->sql.sqlStr)) { return stmtPrepareLiteral2(stmt); } diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index 58bd459f7e3c..38aad10dc219 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); } @@ -6387,6 +6393,8 @@ TEST_F(stmt2CaseF, insert) { TSDB_CODE_SUCCESS, 2), // flaw: redundant (?, ?) shall be reported as error R("insert into stmt2_insert.t (ts, i32) values (?, 1)", TSDB_CODE_SUCCESS, 1), + R("insert into stmt2_insert.? values(?,?)", + TSDB_CODE_SUCCESS, 3), }; #undef R for (size_t i=0; i Date: Wed, 27 May 2026 23:04:23 +0800 Subject: [PATCH 07/16] fix(stmt2): acquireTscObj and releaseTscObj must be paired --- source/client/src/clientStmt2.c | 1 + 1 file changed, 1 insertion(+) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 7d1129148c83..53fb400cf013 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2234,6 +2234,7 @@ static int stmtPrepareLiteral2(TAOS_STMT2* stmt) { 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? } From e22df352f0bfc6643b04ed6fadb34db8128ecb63 Mon Sep 17 00:00:00 2001 From: freemine Date: Thu, 28 May 2026 11:45:49 +0800 Subject: [PATCH 08/16] fix(stmt2): typo correction --- source/client/src/clientStmt2.c | 1 + 1 file changed, 1 insertion(+) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 53fb400cf013..5ffa057ece58 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2331,6 +2331,7 @@ int stmtBindLiteral2(TAOS_STMT2 *stmt) { SET_ERR("no data binding required for literal sql statement"); STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); + return TSDB_CODE_SUCCESS; } static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) { From 7e3bcf2887ac02fc83ce6a432d189fdf016c133c Mon Sep 17 00:00:00 2001 From: freemine Date: Sat, 30 May 2026 07:40:39 +0800 Subject: [PATCH 09/16] fix(stmt2): 1. remove `D` / `A` unused block 2. fully support asynchronous mode triggered by `taos_stmt2_init` 3. `qPureParseInsert` correctly parse VALUES clause 4. legacy support `queryTableNotExistAsEmpty` --- include/libs/parser/parser.h | 28 ---------- source/client/inc/clientStmt2.h | 11 ++-- source/client/src/clientImpl.c | 24 ++++----- source/client/src/clientMain.c | 12 +++-- source/client/src/clientStmt2.c | 90 +++++++++++++++++++++++--------- source/client/test/stmt2Test.cpp | 14 +++++ source/libs/parser/src/parser.c | 39 +++++++++++++- 7 files changed, 143 insertions(+), 75 deletions(-) diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index 428a448c40e5..da8b7c5170e7 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -24,34 +24,6 @@ extern "C" { #include "query.h" #include "querynodes.h" -#include - -#if 0 /* { */ -#if 0 /* { */ -#define D(fmt, ...) \ - fprintf(stderr, "@[%p]%s[%d]:%s():" fmt "\n", \ - (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ - ##__VA_ARGS__) - -#define A(expr, fmt, ...) \ - if (!(expr)) { \ - D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ - abort(); \ - } -#else /* }{ */ -#define D(fmt, ...) \ - syslog(LOG_DEBUG, "@[%p]%s[%d]:%s():" fmt "\n", \ - (void*)(uintptr_t)taosThreadSelf(), __FILE__, __LINE__, __func__, \ - ##__VA_ARGS__) - -#define A(expr, fmt, ...) \ - if (!(expr)) { \ - D("assert `%s` failure:" fmt "", #expr, ##__VA_ARGS__); \ - abort(); \ - } -#endif /* } */ -#endif /* } */ - #define PAR_ERR_RET(c) \ do { \ int32_t _code = c; \ diff --git a/source/client/inc/clientStmt2.h b/source/client/inc/clientStmt2.h index cae6376ab56c..1296a493a3a2 100644 --- a/source/client/inc/clientStmt2.h +++ b/source/client/inc/clientStmt2.h @@ -87,13 +87,14 @@ SSubmitTbData *pCurrTbData; } SStmtExecInfo; */ typedef struct SStmt2LiteralCtx { - tsem_t sem; + tsem_t sem; - int32_t code; + int32_t code; - uint8_t sem_valid:1; - uint8_t prepared:1; - uint8_t executed:1; + 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 } SStmt2LiteralCtx; typedef struct { diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 674ca149edf8..90b8a2a4ab93 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -3463,23 +3463,21 @@ void doRequestCallback(SRequestObj* pRequest, int32_t code) { int64_t this = pRequest->self; - if (!pRequest->literal_by_stmt2) { - 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; - pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT; - if (pRequest->code == TSDB_CODE_PAR_TABLE_NOT_EXIST || pRequest->code == TSDB_CODE_TDB_TABLE_NOT_EXIST) { - pRequest->code = TSDB_CODE_SUCCESS; - if (pRequest->msgBuf != NULL && pRequest->msgBufLen > 0) { - pRequest->msgBuf[0] = '\0'; - } + 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; + pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT; + if (pRequest->code == TSDB_CODE_PAR_TABLE_NOT_EXIST || pRequest->code == TSDB_CODE_TDB_TABLE_NOT_EXIST) { + pRequest->code = TSDB_CODE_SUCCESS; + if (pRequest->msgBuf != NULL && pRequest->msgBufLen > 0) { + pRequest->msgBuf[0] = '\0'; } } - - tscDebug("QID:0x%" PRIx64 ", taos_query end, req:0x%" PRIx64 ", res:%p", pRequest->requestId, pRequest->self, - pRequest); } + tscDebug("QID:0x%" PRIx64 ", taos_query end, req:0x%" PRIx64 ", res:%p", pRequest->requestId, pRequest->self, + pRequest); + if (pRequest->body.queryFp != NULL) { pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code); } diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 9da9c04e6307..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); @@ -2150,8 +2151,13 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { } if (pRequest->literal_by_stmt2) { - doRequestCallback(pRequest, code); - return; + 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); diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 5ffa057ece58..660b9fbd0723 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -27,8 +27,9 @@ char* gStmt2StatusStr[] = {"unknown", "init", "prepare", "settbname", "setta static inline void stmt2LiteralCtxReset(SStmt2LiteralCtx *ctx) { ctx->code = 0; - ctx->prepared = 0; - ctx->executed = 0; + ctx->prepared = 0; + ctx->executing = 0; + ctx->executed = 0; } static inline void @@ -2203,18 +2204,39 @@ static int32_t stmtDeepReset(STscStmt2* pStmt) { 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: internal logic error, not recoverable!!! + // 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(); } + + 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); + } } - pStmt->ctx.code = code; - tsem_post(&pStmt->ctx.sem); } static int stmtPrepareLiteral2(TAOS_STMT2* stmt) { @@ -2330,8 +2352,7 @@ int stmtBindLiteral2(TAOS_STMT2 *stmt) { STscStmt2* pStmt = (STscStmt2*)stmt; SET_ERR("no data binding required for literal sql statement"); - STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); - return TSDB_CODE_SUCCESS; + STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR); } static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) { @@ -3484,32 +3505,47 @@ static int stmtExecLiteral2(TAOS_STMT2* stmt, int *affected_rows) { 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 literal sql statement not supported yet"); + SET_ERR("multiple execution of a prepared literal sql statement " + "not supported yet"); STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); } - pStmt->ctx.executed = 1; + pStmt->ctx.executing = 1; + // NOTE: triggering execution logic of a prepared literal sql statement taosAsyncExecLiteral(stmt); - tsem_wait(&pStmt->ctx.sem); - if (affected_rows) { + 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) { - TAOS_RES *res = pStmt->exec.pRequest; - *affected_rows = taos_affected_rows(res); - } else { - *affected_rows = 0; + if (affected_rows) { + if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { + // NOTE: literal sql statement does not generate any result set + TAOS_RES *res = pStmt->exec.pRequest; + *affected_rows = taos_affected_rows(res); + } else { + // NOTE: literal sql statement generates a result set + *affected_rows = 0; + } + } } + return pStmt->ctx.code; } - if (pStmt->options.asyncExecFn) { - // TODO: a well-defined reentrancy protection is desired, but ... - pStmt->options.asyncExecFn(pStmt->options.userdata, - pStmt->exec.pRequest, pStmt->ctx.code); - } - - return pStmt->ctx.code; + // NOTE: what if taosAsyncExecLiteral failed prematurelly? + return TSDB_CODE_SUCCESS; } int stmtExec2(TAOS_STMT2* stmt, int* affected_rows) { @@ -3731,10 +3767,16 @@ int stmtClose2(TAOS_STMT2* stmt) { const char* stmt2Errstr(TAOS_STMT2* stmt) { STscStmt2* pStmt = (STscStmt2*)stmt; - if (stmt == NULL || NULL == pStmt->exec.pRequest) { - if (stmt && stmtIsLiteral(pStmt)) { + 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; } + return tstrerror(pStmt->ctx.code); + } + + if (stmt == NULL || NULL == pStmt->exec.pRequest) { return (char*)tstrerror(terrno); } diff --git a/source/client/test/stmt2Test.cpp b/source/client/test/stmt2Test.cpp index 38aad10dc219..dbaba68787a1 100644 --- a/source/client/test/stmt2Test.cpp +++ b/source/client/test/stmt2Test.cpp @@ -6395,6 +6395,20 @@ TEST_F(stmt2CaseF, insert) { TSDB_CODE_SUCCESS, 1), R("insert into stmt2_insert.? values(?,?)", TSDB_CODE_SUCCESS, 3), + R("insert into stmt2_insert.? values(now(),?)", + TSDB_CODE_SUCCESS, 2), + R("insert into stmt2_insert.? values(now,?)", + TSDB_CODE_SUCCESS, 2), + R("insert into stmt2_insert.? values(now,null)", + TSDB_CODE_SUCCESS, 1), + R("insert into stmt2_insert.? values(now,true)", + TSDB_CODE_SUCCESS, 1), + R("insert into stmt2_insert.? values(now,false)", + TSDB_CODE_SUCCESS, 1), + R("insert into stmt2_insert.? values(now,-1)", + TSDB_CODE_SUCCESS, 1), + R("insert into stmt2_insert.? values(now,-1.2)", + TSDB_CODE_SUCCESS, 1), }; #undef R for (size_t i=0; inr_params; ++nr_values; + ADVANCE(); } else if (MATCH(TK_NK_INTEGER) || MATCH(TK_NK_FLOAT) - || MATCH(TK_NK_STRING)) { + || 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 "); } - ADVANCE(); if (MATCH(TK_NK_COMMA)) { if (nr_values == nr_names) { RETURN_EXPECTING("`)`"); From d7072c7b63ca0304285658ee10a31ad80e7d6cc4 Mon Sep 17 00:00:00 2001 From: freemine Date: Sat, 30 May 2026 08:19:08 +0800 Subject: [PATCH 10/16] fix(stmt2): for `QueryTbNotExistAsEmpty` + table_not_exists case, keep no-fields-result-set mandatorily --- source/client/src/clientStmt2.c | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 660b9fbd0723..92c74e4c1554 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -3914,7 +3914,11 @@ TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) { pStmt->errCode = TSDB_CODE_TSC_STMT_API_ERROR; return NULL; } - if (taos_num_fields(pStmt->exec.pRequest) == 0) { + if (pStmt->exec.pRequest && + pStmt->exec.pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT) { + // NOTE: empty result when `QueryTbNotExistAsEmpty` is set + // and table not exists + } else if (taos_num_fields(pStmt->exec.pRequest) == 0) { STMT2_ELOG_E("useResult only for query statement even it's literal"); return NULL; } From 581d339fe6d3c87f6609ac9750d3ea3b8d10c72c Mon Sep 17 00:00:00 2001 From: freemine Date: Sat, 30 May 2026 19:32:05 +0800 Subject: [PATCH 11/16] fix(stmt2): distinguish where literal statement generates result set by stmt2 APIs --- source/client/inc/clientStmt2.h | 13 +++++------ source/client/src/clientStmt2.c | 38 ++++++++++++++++++++++----------- 2 files changed, 32 insertions(+), 19 deletions(-) diff --git a/source/client/inc/clientStmt2.h b/source/client/inc/clientStmt2.h index 1296a493a3a2..c91ac362f3d0 100644 --- a/source/client/inc/clientStmt2.h +++ b/source/client/inc/clientStmt2.h @@ -87,14 +87,15 @@ SSubmitTbData *pCurrTbData; } SStmtExecInfo; */ typedef struct SStmt2LiteralCtx { - tsem_t sem; + tsem_t sem; - int32_t code; + 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 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 { diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 92c74e4c1554..72beb8415440 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -27,9 +27,10 @@ char* gStmt2StatusStr[] = {"unknown", "init", "prepare", "settbname", "setta static inline void stmt2LiteralCtxReset(SStmt2LiteralCtx *ctx) { ctx->code = 0; - ctx->prepared = 0; - ctx->executing = 0; - ctx->executed = 0; + ctx->prepared = 0; + ctx->executing = 0; + ctx->executed = 0; + ctx->has_result_set = 0; } static inline void @@ -3530,14 +3531,24 @@ static int stmtExecLiteral2(TAOS_STMT2* stmt, int *affected_rows) { pStmt->ctx.executed = 1; if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { + 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 (affected_rows) { - if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { + 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); - } else { - // NOTE: literal sql statement generates a result set - *affected_rows = 0; } } } @@ -3909,17 +3920,18 @@ TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) { STMT2_TLOG_E("start to use result"); if (stmtIsLiteral(pStmt)) { + if (pStmt->ctx.executing == 0) { + 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->exec.pRequest && - pStmt->exec.pRequest->type == TSDB_SQL_RETRIEVE_EMPTY_RESULT) { - // NOTE: empty result when `QueryTbNotExistAsEmpty` is set - // and table not exists - } else if (taos_num_fields(pStmt->exec.pRequest) == 0) { - STMT2_ELOG_E("useResult only for query statement even it's literal"); + if (!pStmt->ctx.has_result_set) { + STMT2_ELOG_E("useResult only for query statement"); return NULL; } return pStmt->exec.pRequest; From fc237dee6e81ac946728befca7b99cf736eab080 Mon Sep 17 00:00:00 2001 From: freemine Date: Sat, 30 May 2026 22:26:47 +0800 Subject: [PATCH 12/16] fix(stmt2): has_result_set shall be set in callback function --- source/client/src/clientStmt2.c | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 72beb8415440..6a996952939c 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2223,6 +2223,17 @@ static void stmtLiteralCallback(void *param, TAOS_RES *res, int code) { 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 ... @@ -3531,16 +3542,6 @@ static int stmtExecLiteral2(TAOS_STMT2* stmt, int *affected_rows) { pStmt->ctx.executed = 1; if (pStmt->ctx.code == TSDB_CODE_SUCCESS) { - 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 (affected_rows) { if (pStmt->ctx.has_result_set) { // NOTE: literal sql statement generates a result set @@ -3920,7 +3921,7 @@ TAOS_RES* stmtUseResult2(TAOS_STMT2* stmt) { STMT2_TLOG_E("start to use result"); if (stmtIsLiteral(pStmt)) { - if (pStmt->ctx.executing == 0) { + if (pStmt->ctx.executing) { SET_ERR("literal sql statement still in progress"); pStmt->errCode = TSDB_CODE_TSC_STMT_API_ERROR; return NULL; From 6beea071e5bf729536dbb61ecc8dad5876689727 Mon Sep 17 00:00:00 2001 From: freemine Date: Tue, 2 Jun 2026 08:17:13 +0800 Subject: [PATCH 13/16] fix(stmt2): reuse pRequest->msgBuf if available --- source/client/src/clientStmt2.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 6a996952939c..ebcadcd280e7 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -3785,7 +3785,7 @@ const char* stmt2Errstr(TAOS_STMT2* stmt) { // error msg is stored in pStmt->msgBuf via `SET_ERR` return pStmt->msgBuf; } - return tstrerror(pStmt->ctx.code); + return pStmt->exec.pRequest->msgBuf; } if (stmt == NULL || NULL == pStmt->exec.pRequest) { From 462e523e501a82726d1ceef47a1f61db07e7365d Mon Sep 17 00:00:00 2001 From: freemine Date: Wed, 3 Jun 2026 07:29:17 +0800 Subject: [PATCH 14/16] fix(stmt2): 1. reuse SRequestObj::msgBuf if possible 2. SRequestObj lifecycle management adjustment in `clientImpl.c::doRequestCallback` --- source/client/src/clientImpl.c | 17 ++++++++--------- source/client/src/clientStmt2.c | 3 +++ 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 90b8a2a4ab93..7f694d6ea400 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -3462,6 +3462,12 @@ 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)) { @@ -3482,15 +3488,8 @@ void doRequestCallback(SRequestObj* pRequest, int32_t code) { pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code); } - if (!pRequest->literal_by_stmt2) { - SRequestObj* pReq = acquireRequest(this); - if (pReq != NULL) { - pReq->inCallback = false; - (void)releaseRequest(this); - } - } else { - pRequest->inCallback = false; - } + 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/clientStmt2.c b/source/client/src/clientStmt2.c index ebcadcd280e7..dd086e338893 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -3784,6 +3784,9 @@ const char* stmt2Errstr(TAOS_STMT2* stmt) { // 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 pStmt->exec.pRequest->msgBuf; } From fe17b3ef8884d923d684b41a588357197a6618d5 Mon Sep 17 00:00:00 2001 From: freemine Date: Wed, 3 Jun 2026 18:00:09 +0800 Subject: [PATCH 15/16] fix(stmt2): does NOT reset/release stmt2LiteralCtx until asynchronous operations have completed --- source/client/src/clientStmt2.c | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index dd086e338893..840f09563634 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -2122,9 +2122,6 @@ static int32_t stmtDeepReset(STscStmt2* pStmt) { pStmt->errCode = 0; - stmt2LiteralCtxReset(&pStmt->ctx); - pStmt->literal = 0; - // Wait for async execution to complete if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) { if (tsem_wait(&pStmt->asyncExecSem) != 0) { @@ -2153,6 +2150,10 @@ 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: @@ -3733,8 +3734,6 @@ int stmtClose2(TAOS_STMT2* stmt) { (void)taosThreadCondDestroy(&pStmt->asyncBindParam.waitCond); (void)taosThreadMutexDestroy(&pStmt->asyncBindParam.mutex); - stmt2LiteralCtxRelease(&pStmt->ctx); - if (pStmt->options.asyncExecFn && !pStmt->execSemWaited) { if (tsem_wait(&pStmt->asyncExecSem) != 0) { STMT2_ELOG_E("fail to wait asyncExecSem"); @@ -3749,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" From f0ea74dd3cd693f477324ed19774899cea5095a4 Mon Sep 17 00:00:00 2001 From: freemine Date: Fri, 5 Jun 2026 12:22:57 +0800 Subject: [PATCH 16/16] fix(stmt2): typo correction --- source/client/src/clientStmt2.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/client/src/clientStmt2.c b/source/client/src/clientStmt2.c index 840f09563634..83f759f1717a 100644 --- a/source/client/src/clientStmt2.c +++ b/source/client/src/clientStmt2.c @@ -3790,7 +3790,7 @@ const char* stmt2Errstr(TAOS_STMT2* stmt) { // NOTE: reuse error msg stored in `msgBuf` return pStmt->exec.pRequest->msgBuf; } - return pStmt->exec.pRequest->msgBuf; + return (const char*)tstrerror(pStmt->exec.pRequest->code); } if (stmt == NULL || NULL == pStmt->exec.pRequest) {