Skip to content

Refactor/freemine.tdengine.stmt2 #35366

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
freemine wants to merge 16 commits into
base: main
Choose a base branch
from
Open

Refactor/freemine.tdengine.stmt2 #35366

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
332f432
feat(stmt2): literal sql statement can be exec-direct, that is taos_s…
freemine May 25, 2026
60e5796
feat(stmt2): fully prepared for most insert statement, except for ins…
freemine May 25, 2026
b046501
feat(stmt2): literal sql statement can be exec-direct in an async mode
freemine May 26, 2026
dc9afb5
feat(stmt2): insert into [db.]? (...) values (...) also fully prepare…
freemine May 26, 2026
9729efd
fix(stmt2): several tiny bugs/typos found in parameterized-literal-pr…
freemine May 27, 2026
3004f93
fix(stmt2): adjust stmt2Test test cases accordingly, and some tiny bu…
freemine May 27, 2026
fe056b8
fix(stmt2): acquireTscObj and releaseTscObj must be paired
freemine May 27, 2026
e22df35
fix(stmt2): typo correction
freemine May 28, 2026
7e3bcf2
fix(stmt2):
freemine May 29, 2026
d7072c7
fix(stmt2): for `QueryTbNotExistAsEmpty` + table_not_exists case, kee…
freemine May 30, 2026
581d339
fix(stmt2): distinguish where literal statement generates result set …
freemine May 30, 2026
fc237de
fix(stmt2): has_result_set shall be set in callback function
freemine May 30, 2026
6beea07
fix(stmt2): reuse pRequest->msgBuf if available
freemine Jun 2, 2026
462e523
fix(stmt2):
freemine Jun 2, 2026
fe17b3e
fix(stmt2): does NOT reset/release stmt2LiteralCtx until asynchronous…
freemine Jun 3, 2026
f0ea74d
fix(stmt2): typo correction
freemine Jun 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions include/libs/parser/parser.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
6 changes: 5 additions & 1 deletion source/client/inc/clientInt.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
freemine marked this conversation as resolved.
} SRequestObj;

typedef struct SSyncQueryParam {
Expand All @@ -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,
Expand Down
20 changes: 20 additions & 0 deletions source/client/inc/clientStmt2.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -176,6 +188,10 @@ typedef struct {
bool asyncResultAvailable;
SStmtStatInfo stat;
SArray* pVgDataBlocksForRetry; // SArray<SVgDataBlocks*> saved serialized data for NEED_CLIENT_HANDLE_ERROR retry

char msgBuf[128];
SStmt2LiteralCtx ctx;
uint8_t literal:1;
} STscStmt2;
/*
extern char *gStmtStatusStr[];
Expand Down Expand Up @@ -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);
Expand All @@ -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
Expand Down
20 changes: 13 additions & 7 deletions source/client/src/clientImpl.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,

@Pengrongkun Pengrongkun May 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

stmt的函数单独封装，不要和query耦合在一起

@freemine freemine May 29, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

just don't get a clear solutions, otherwise i have to hard-copy the whole logic in the block.
or, what suggestions might be here?

int64_t reqid) {
if (sql == NULL || NULL == fp) {
terrno = TSDB_CODE_INVALID_PARA;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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(&param->sem);
if (TSDB_CODE_SUCCESS != code) {
taosMemoryFree(param);
Expand Down Expand Up @@ -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;
Expand All @@ -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,
Expand Down
77 changes: 57 additions & 20 deletions source/client/src/clientMain.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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) {

@Pengrongkun Pengrongkun Jun 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

问题：SRequestObj.literal_by_stmt2 是指向 TAOS_STMT2* 的裸指针（注释为 // reference only），无所有权/生命周期协议。在异步 exec 路径中，doAsyncQuery 会解引用该指针检查 ctx.prepared。若 stmt 在请求飞行期间被 close，该指针即变为悬空指针。

证据：

  • clientInt.h：TAOS_STMT2 *literal_by_stmt2; // reference only
  • clientMain.c:doAsyncQuery：TAOS_STMT2* stmt = pRequest->literal_by_stmt2; STscStmt2* pStmt = (STscStmt2*)stmt;

与本 MR 的关联：本 MR 引入该请求→stmt 反向引用以实现 literal exec-direct 功能。

修复建议：在赋值 literal_by_stmt2 时对 stmt 加引用计数，请求完成后释放；或在 stmt reset/close 时显式将 pRequest->literal_by_stmt2 置 NULL（需要加锁）。

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;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
Loading
Loading