From be021333116fe67cd10c9dbc735041d44029f35f Mon Sep 17 00:00:00 2001 From: Hui Li Date: Wed, 3 Jun 2020 12:15:29 +0800 Subject: [PATCH] [TD-489] --- src/client/inc/tsclient.h | 2 + src/client/src/tscAsync.c | 8 +++ src/client/src/tscParseInsert.c | 3 +- src/client/src/tscPrepare.c | 105 ++++++++++++++++++++++---------- src/client/src/tscSql.c | 2 +- src/inc/taosdef.h | 1 + 6 files changed, 86 insertions(+), 35 deletions(-) diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index f1b620176d..3c1d43d688 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -317,6 +317,7 @@ typedef struct SSqlObj { SRpcIpSet ipList; char freed : 4; char listed : 4; + uint32_t insertType; tsem_t rspSem; SSqlCmd cmd; SSqlRes res; @@ -402,6 +403,7 @@ void tscCloseTscObj(STscObj *pObj); TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port, void (*fp)(void *, TAOS_RES *, int), void *param, void **taos); +void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ; void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen); diff --git a/src/client/src/tscAsync.c b/src/client/src/tscAsync.c index 96837e4dd4..18f9872160 100644 --- a/src/client/src/tscAsync.c +++ b/src/client/src/tscAsync.c @@ -482,6 +482,14 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) { } } else { code = tsParseSql(pSql, false); + if ((pQueryInfo->type & TSDB_QUERY_TYPE_STMT_INSERT) == TSDB_QUERY_TYPE_STMT_INSERT) { + STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); + code = tscGetTableMeta(pSql, pTableMetaInfo); + assert(code == TSDB_CODE_SUCCESS && pTableMetaInfo->pTableMeta != NULL); + (*pSql->fp)(pSql->param, NULL, code); + return; + } + if (code == TSDB_CODE_ACTION_IN_PROGRESS) return; } } diff --git a/src/client/src/tscParseInsert.c b/src/client/src/tscParseInsert.c index 79872e22c8..d914f392fd 100644 --- a/src/client/src/tscParseInsert.c +++ b/src/client/src/tscParseInsert.c @@ -1312,6 +1312,7 @@ int tsParseInsertSql(SSqlObj *pSql) { tscGetQueryInfoDetailSafely(pCmd, pCmd->clauseIndex, &pQueryInfo); TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT); + TSDB_QUERY_SET_TYPE(pQueryInfo->type, pSql->insertType); sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL); if (sToken.type != TK_INTO) { @@ -1339,7 +1340,7 @@ int tsParseSql(SSqlObj *pSql, bool initialParse) { * Set the fp before parse the sql string, in case of getTableMeta failed, in which * the error handle callback function can rightfully restore the user-defined callback function (fp). */ - if (initialParse) { + if (initialParse && (pSql->insertType != TSDB_QUERY_TYPE_STMT_INSERT)) { pSql->fetchFp = pSql->fp; pSql->fp = (void(*)())tscHandleMultivnodeInsert; } diff --git a/src/client/src/tscPrepare.c b/src/client/src/tscPrepare.c index 5aaa53c549..d66e40b20f 100644 --- a/src/client/src/tscPrepare.c +++ b/src/client/src/tscPrepare.c @@ -12,6 +12,7 @@ * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ +#include "os.h" #include "taos.h" #include "tsclient.h" @@ -20,6 +21,7 @@ #include "taosmsg.h" #include "tstrbuild.h" #include "tscLog.h" +#include "tscSubquery.h" int tsParseInsertSql(SSqlObj *pSql); int taos_query_imp(STscObj* pObj, SSqlObj* pSql); @@ -262,7 +264,11 @@ static char* normalStmtBuildSql(STscStmt* stmt) { static int doBindParam(char* data, SParamInfo* param, TAOS_BIND* bind) { if (bind->is_null != NULL && *(bind->is_null)) { - setNull(data, param->type, param->bytes); + if (param->type == TSDB_DATA_TYPE_BINARY || param->type == TSDB_DATA_TYPE_NCHAR) { + setVardataNull(data + param->offset, param->type); + } else { + setNull(data + param->offset, param->type, param->bytes); + } return TSDB_CODE_SUCCESS; } @@ -297,14 +303,17 @@ static int doBindParam(char* data, SParamInfo* param, TAOS_BIND* bind) { return TSDB_CODE_INVALID_VALUE; } size = (short)*bind->length; - break; + STR_WITH_SIZE_TO_VARSTR(data + param->offset, bind->buffer, size); + return TSDB_CODE_SUCCESS; - case TSDB_DATA_TYPE_NCHAR: - if (!taosMbsToUcs4(bind->buffer, *bind->length, data + param->offset, param->bytes, NULL)) { + case TSDB_DATA_TYPE_NCHAR: { + size_t output = 0; + if (!taosMbsToUcs4(bind->buffer, *bind->length, varDataVal(data + param->offset), param->bytes - VARSTR_HEADER_SIZE, &output)) { return TSDB_CODE_INVALID_VALUE; - } + } + varDataSetLen(data + param->offset, output); return TSDB_CODE_SUCCESS; - + } default: assert(false); return TSDB_CODE_INVALID_VALUE; @@ -383,14 +392,6 @@ static int insertStmtAddBatch(STscStmt* stmt) { return TSDB_CODE_SUCCESS; } -static int insertStmtPrepare(STscStmt* stmt) { - SSqlObj *pSql = stmt->pSql; - pSql->cmd.numOfParams = 0; - pSql->cmd.batchSize = 0; - - return tsParseInsertSql(pSql); -} - static int insertStmtReset(STscStmt* pStmt) { SSqlCmd* pCmd = &pStmt->pSql->cmd; if (pCmd->batchSize > 2) { @@ -451,14 +452,16 @@ static int insertStmtExecute(STscStmt* stmt) { pRes->qhandle = 0; + pSql->insertType = 0; + pSql->fetchFp = waitForQueryRsp; + pSql->fp = (void(*)())tscHandleMultivnodeInsert; + tscDoQuery(pSql); - // tscTrace("%p SQL result:%d, %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj); - if (pRes->code != TSDB_CODE_SUCCESS) { - tscPartiallyFreeSqlObj(pSql); - } + // wait for the callback function to post the semaphore + tsem_wait(&pSql->rspSem); + return pSql->res.code; - return pRes->code; } //////////////////////////////////////////////////////////////////////////////// @@ -478,6 +481,7 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) { tscError("failed to allocate memory for statement"); return NULL; } + pStmt->taos = pObj; SSqlObj* pSql = calloc(1, sizeof(SSqlObj)); if (pSql == NULL) { @@ -488,8 +492,10 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) { } tsem_init(&pSql->rspSem, 0, 0); - pSql->signature = pSql; - pSql->pTscObj = pObj; + pSql->signature = pSql; + pSql->pTscObj = pObj; + pSql->pTscObj->pSql = pSql; + pSql->maxRetry = TSDB_MAX_REPLICA_NUM; pStmt->pSql = pSql; return pStmt; @@ -497,22 +503,55 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) { int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) { STscStmt* pStmt = (STscStmt*)stmt; - if (length == 0) { - length = strlen(sql); + + if (stmt == NULL || pStmt->taos == NULL || pStmt->pSql == NULL) { + terrno = TSDB_CODE_DISCONNECTED; + return TSDB_CODE_DISCONNECTED; + } + + SSqlObj* pSql = pStmt->pSql; + size_t sqlLen = strlen(sql); + + //doAsyncQuery(pObj, pSql, waitForQueryRsp, taos, sqlstr, sqlLen); + SSqlCmd *pCmd = &pSql->cmd; + SSqlRes *pRes = &pSql->res; + pSql->param = (void*)pStmt->taos; + pSql->fp = waitForQueryRsp; + pSql->insertType = TSDB_QUERY_TYPE_STMT_INSERT; + + if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) { + tscError("%p failed to malloc payload buffer", pSql); + return TSDB_CODE_CLI_OUT_OF_MEMORY; } - char* sqlstr = (char*)malloc(length + 1); - if (sqlstr == NULL) { - tscError("failed to malloc sql string buffer"); + + pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1); + + if (pSql->sqlstr == NULL) { + tscError("%p failed to malloc sql string buffer", pSql); + free(pCmd->payload); return TSDB_CODE_CLI_OUT_OF_MEMORY; } - memcpy(sqlstr, sql, length); - sqlstr[length] = 0; - strtolower(sqlstr, sqlstr); + + pRes->qhandle = 0; + pRes->numOfRows = 1; + + strtolower(pSql->sqlstr, sql); + tscDump("%p SQL: %s", pSql, pSql->sqlstr); - pStmt->pSql->sqlstr = sqlstr; - if (tscIsInsertData(sqlstr)) { + if (tscIsInsertData(pSql->sqlstr)) { pStmt->isInsert = true; - return insertStmtPrepare(pStmt); + + pSql->cmd.numOfParams = 0; + pSql->cmd.batchSize = 0; + + int32_t code = tsParseSql(pSql, true); + if (code == TSDB_CODE_ACTION_IN_PROGRESS) { + // wait for the callback function to post the semaphore + tsem_wait(&pSql->rspSem); + return pSql->res.code; + } + + return code; } pStmt->isInsert = false; @@ -574,7 +613,7 @@ int taos_stmt_execute(TAOS_STMT* stmt) { } else { tfree(pStmt->pSql->sqlstr); pStmt->pSql->sqlstr = sql; - ret = taos_query_imp(pStmt->taos, pStmt->pSql); + ret = taos_query(pStmt->taos, pStmt->pSql->sqlstr); } } return ret; diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index a4cbd7f7ec..a8e4a077ea 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -264,7 +264,7 @@ int taos_query_imp(STscObj *pObj, SSqlObj *pSql) { return pRes->code; } -static void waitForQueryRsp(void *param, TAOS_RES *tres, int code) { +void waitForQueryRsp(void *param, TAOS_RES *tres, int code) { assert(param != NULL); SSqlObj *pSql = ((STscObj *)param)->pSql; diff --git a/src/inc/taosdef.h b/src/inc/taosdef.h index 2aca057ba7..7ec9aef295 100644 --- a/src/inc/taosdef.h +++ b/src/inc/taosdef.h @@ -331,6 +331,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size); #define TSDB_QUERY_TYPE_TAG_FILTER_QUERY 0x400u #define TSDB_QUERY_TYPE_INSERT 0x100u // insert type #define TSDB_QUERY_TYPE_MULTITABLE_QUERY 0x200u +#define TSDB_QUERY_TYPE_STMT_INSERT 0x800u // stmt insert type #define TSDB_QUERY_HAS_TYPE(x, _type) (((x) & (_type)) != 0) #define TSDB_QUERY_SET_TYPE(x, _type) ((x) |= (_type)) -- GitLab