xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

stmt

This commit is contained in:
dapan1121 2022-04-18 19:08:27 +08:00
parent 49e0ecdb9b
commit 0c2f22fb82
8 changed files with 180 additions and 82 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/libs/parser/parser.h
View File

@ -82,7 +82,7 @@ void qDestroyQuery(SQuery* pQueryNode);
int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema); int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash); int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
void qResetStmtDataBlock(void* pBlock, bool freeData); int32_t qResetStmtDataBlock(void* block, bool keepBuf);
int32_t qCloneStmtDataBlock(void** pDst, void* pSrc); int32_t qCloneStmtDataBlock(void** pDst, void* pSrc);
void qFreeStmtDataBlock(void* pDataBlock); void qFreeStmtDataBlock(void* pDataBlock);
int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc); int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc);

20
source/client/inc/clientStmt.h
View File

@ -33,6 +33,7 @@ typedef enum {
STMT_INIT = 1, STMT_INIT = 1,
STMT_PREPARE, STMT_PREPARE,
STMT_SETTBNAME, STMT_SETTBNAME,
STMT_SETTAGS,
STMT_FETCH_TAG_FIELDS, STMT_FETCH_TAG_FIELDS,
STMT_FETCH_COL_FIELDS, STMT_FETCH_COL_FIELDS,
STMT_BIND, STMT_BIND,
@ -87,25 +88,6 @@ typedef struct STscStmt {
#define STMT_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0) #define STMT_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define STMT_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0) #define STMT_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
#define STMT_SWITCH_STATUS(_stmt, _newstatus, _errcode) \
do { \
switch (_newstatus) { \
case STMT_INIT: \
break; \
case STMT_PREPARE: \
if ((_stmt)->sql.status != STMT_INIT) STMT_ERR_RET(_errcode); \
break; \
case STMT_SETTBNAME: \
break; \
default: \
STMT_ERR_RET(_errcode); \
break; \
} \
\
(_stmt)->sql.status = _newstatus; \
} while (0)
TAOS_STMT *stmtInit(TAOS *taos); TAOS_STMT *stmtInit(TAOS *taos);
int stmtClose(TAOS_STMT *stmt); int stmtClose(TAOS_STMT *stmt);
int stmtExec(TAOS_STMT *stmt); int stmtExec(TAOS_STMT *stmt);

8
source/client/src/clientMain.c
View File

@ -655,7 +655,7 @@ int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind) {
} }
int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind, int colIdx) { int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind, int colIdx) {
return stmtBindBatch(stmt, bind); /* TODO */ return stmtBindBatch(stmt, bind, colIdx); /* TODO */
} }
int taos_stmt_add_batch(TAOS_STMT *stmt) { int taos_stmt_add_batch(TAOS_STMT *stmt) {
@ -709,12 +709,6 @@ TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt) {
} }
char *taos_stmt_errstr(TAOS_STMT *stmt) { char *taos_stmt_errstr(TAOS_STMT *stmt) {
if (stmt == NULL) {
tscError("NULL parameter for %s", __FUNCTION__);
terrno = TSDB_CODE_INVALID_PARA;
return NULL;
}
return (char *)stmtErrstr(stmt); return (char *)stmtErrstr(stmt);
} }

74
source/client/src/clientStmt.c
View File

@ -4,6 +4,22 @@
#include "clientStmt.h" #include "clientStmt.h"
#include "tdef.h" #include "tdef.h"
int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
switch (newStatus) {
case STMT_SETTBNAME:
break;
default:
break;
}
//STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
pStmt->sql.status = newStatus;
return TSDB_CODE_SUCCESS;
}
int32_t stmtGetTbName(TAOS_STMT *stmt, char **tbName) { int32_t stmtGetTbName(TAOS_STMT *stmt, char **tbName) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
@ -115,6 +131,7 @@ int32_t stmtParseSql(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t stmtCleanBindInfo(STscStmt* pStmt) { int32_t stmtCleanBindInfo(STscStmt* pStmt) {
pStmt->bInfo.tbUid = 0; pStmt->bInfo.tbUid = 0;
pStmt->bInfo.tbSuid = 0; pStmt->bInfo.tbSuid = 0;
@ -138,7 +155,7 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable) {
uint64_t *key = taosHashGetKey(pIter, NULL); uint64_t *key = taosHashGetKey(pIter, NULL);
if (keepTable && (*key == pStmt->bInfo.tbUid)) { if (keepTable && (*key == pStmt->bInfo.tbUid)) {
qResetStmtDataBlock(pBlocks, true); STMT_ERR_RET(qResetStmtDataBlock(pBlocks, true));
pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter); pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
continue; continue;
@ -187,6 +204,8 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
} }
int32_t stmtGetFromCache(STscStmt* pStmt) { int32_t stmtGetFromCache(STscStmt* pStmt) {
pStmt->bInfo.needParse = true;
if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) { if (NULL == pStmt->sql.pTableCache || taosHashGetSize(pStmt->sql.pTableCache) <= 0) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -241,6 +260,8 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
STMT_ERR_RET(stmtCleanBindInfo(pStmt));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -276,7 +297,11 @@ int stmtPrepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
STMT_ERR_RET(stmtCleanSQLInfo(pStmt)); STMT_ERR_RET(stmtCleanSQLInfo(pStmt));
} }
STMT_SWITCH_STATUS(pStmt, STMT_PREPARE, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_PREPARE));
if (length <= 0) {
length = strlen(sql);
}
pStmt->sql.sqlStr = strndup(sql, length); pStmt->sql.sqlStr = strndup(sql, length);
pStmt->sql.sqlLen = length; pStmt->sql.sqlLen = length;
@ -288,39 +313,40 @@ int stmtPrepare(TAOS_STMT *stmt, const char *sql, unsigned long length) {
int stmtSetTbName(TAOS_STMT *stmt, const char *tbName) { int stmtSetTbName(TAOS_STMT *stmt, const char *tbName) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_SWITCH_STATUS(pStmt, STMT_SETTBNAME, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTBNAME));
taosMemoryFree(pStmt->bInfo.tbName);
if (NULL == pStmt->exec.pRequest) { if (NULL == pStmt->exec.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest)); STMT_ERR_RET(buildRequest(pStmt->taos, pStmt->sql.sqlStr, pStmt->sql.sqlLen, &pStmt->exec.pRequest));
} }
STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen)); STMT_ERR_RET(qCreateSName(&pStmt->bInfo.sname, tbName, pStmt->taos->acctId, pStmt->exec.pRequest->pDb, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
pStmt->bInfo.tbName = strdup(tbName);
STMT_ERR_RET(stmtGetFromCache(pStmt)); STMT_ERR_RET(stmtGetFromCache(pStmt));
if (pStmt->bInfo.needParse) {
taosMemoryFree(pStmt->bInfo.tbName);
pStmt->bInfo.tbName = strdup(tbName);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int stmtSetTbTags(TAOS_STMT *stmt, TAOS_BIND_v2 *tags) { int stmtSetTbTags(TAOS_STMT *stmt, TAOS_BIND_v2 *tags) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_SWITCH_STATUS(pStmt, STMT_SETTBNAME, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
if (pStmt->bInfo.needParse) { if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt)); STMT_ERR_RET(stmtParseSql(pStmt));
} }
STableDataBlocks *pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid)); STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid); tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
} }
STMT_ERR_RET(qBindStmtTagsValue(pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, &pStmt->bInfo.sname, tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen)); STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, &pStmt->bInfo.sname, tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -329,7 +355,7 @@ int stmtSetTbTags(TAOS_STMT *stmt, TAOS_BIND_v2 *tags) {
int32_t stmtFetchTagFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fields) { int32_t stmtFetchTagFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fields) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_SWITCH_STATUS(pStmt, STMT_FETCH_TAG_FIELDS, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_TAG_FIELDS));
if (pStmt->bInfo.needParse) { if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt)); STMT_ERR_RET(stmtParseSql(pStmt));
@ -340,13 +366,13 @@ int32_t stmtFetchTagFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fiel
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
} }
STableDataBlocks *pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid)); STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid); tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
} }
STMT_ERR_RET(qBuildStmtTagFields(pDataBlock, pStmt->bInfo.boundTags, fieldNum, fields)); STMT_ERR_RET(qBuildStmtTagFields(*pDataBlock, pStmt->bInfo.boundTags, fieldNum, fields));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -354,7 +380,7 @@ int32_t stmtFetchTagFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fiel
int32_t stmtFetchColFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fields) { int32_t stmtFetchColFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fields) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_SWITCH_STATUS(pStmt, STMT_FETCH_COL_FIELDS, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_COL_FIELDS));
if (pStmt->bInfo.needParse) { if (pStmt->bInfo.needParse) {
STMT_ERR_RET(stmtParseSql(pStmt)); STMT_ERR_RET(stmtParseSql(pStmt));
@ -365,13 +391,13 @@ int32_t stmtFetchColFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fiel
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
} }
STableDataBlocks *pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid)); STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid); tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
} }
STMT_ERR_RET(qBuildStmtColFields(pDataBlock, fieldNum, fields)); STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -379,7 +405,7 @@ int32_t stmtFetchColFields(TAOS_STMT *stmt, int32_t *fieldNum, TAOS_FIELD** fiel
int stmtBindBatch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind) { int stmtBindBatch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_SWITCH_STATUS(pStmt, STMT_BIND, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) { if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
pStmt->bInfo.needParse = false; pStmt->bInfo.needParse = false;
@ -393,13 +419,13 @@ int stmtBindBatch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind) {
STMT_ERR_RET(stmtParseSql(pStmt)); STMT_ERR_RET(stmtParseSql(pStmt));
} }
STableDataBlocks *pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid)); STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
if (NULL == pDataBlock) { if (NULL == pDataBlock) {
tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid); tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR); STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
} }
qBindStmtColsValue(pDataBlock, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen); qBindStmtColsValue(*pDataBlock, bind, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -408,7 +434,7 @@ int stmtBindBatch(TAOS_STMT *stmt, TAOS_BIND_v2 *bind) {
int stmtAddBatch(TAOS_STMT *stmt) { int stmtAddBatch(TAOS_STMT *stmt) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
STMT_SWITCH_STATUS(pStmt, STMT_ADD_BATCH, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
STMT_ERR_RET(stmtCacheBlock(pStmt)); STMT_ERR_RET(stmtCacheBlock(pStmt));
@ -419,7 +445,7 @@ int stmtExec(TAOS_STMT *stmt) {
STscStmt* pStmt = (STscStmt*)stmt; STscStmt* pStmt = (STscStmt*)stmt;
int32_t code = 0; int32_t code = 0;
STMT_SWITCH_STATUS(pStmt, STMT_EXECUTE, TSDB_CODE_TSC_STMT_API_ERROR); STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_EXECUTE));
STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash)); STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
@ -448,6 +474,10 @@ const char *stmtErrstr(TAOS_STMT *stmt) {
return (char*) tstrerror(terrno); return (char*) tstrerror(terrno);
} }
if (pStmt->exec.pRequest) {
pStmt->exec.pRequest->code = terrno;
}
return taos_errstr(pStmt->exec.pRequest); return taos_errstr(pStmt->exec.pRequest);
} }

26
source/libs/parser/src/parInsert.c
View File

@ -885,7 +885,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken)
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, int16_t timePrec, int32_t* len, char* tmpTokenBuf) { static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, int16_t timePrec, bool* gotRow, char* tmpTokenBuf) {
SParsedDataColInfo* spd = &pDataBlocks->boundColumnInfo; SParsedDataColInfo* spd = &pDataBlocks->boundColumnInfo;
SRowBuilder* pBuilder = &pDataBlocks->rowBuilder; SRowBuilder* pBuilder = &pDataBlocks->rowBuilder;
STSRow* row = (STSRow*)(pDataBlocks->pData + pDataBlocks->size); // skip the SSubmitBlk header STSRow* row = (STSRow*)(pDataBlocks->pData + pDataBlocks->size); // skip the SSubmitBlk header
@ -937,6 +937,8 @@ static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks,
} }
} }
} }
*gotRow = true;
} }
// *len = pBuilder->extendedRowSize; // *len = pBuilder->extendedRowSize;
@ -967,19 +969,23 @@ static int32_t parseValues(SInsertParseContext* pCxt, STableDataBlocks* pDataBlo
maxRows = tSize; maxRows = tSize;
} }
int32_t len = 0; bool gotRow = false;
CHECK_CODE(parseOneRow(pCxt, pDataBlock, tinfo.precision, &len, tmpTokenBuf)); CHECK_CODE(parseOneRow(pCxt, pDataBlock, tinfo.precision, &gotRow, tmpTokenBuf));
pDataBlock->size += extendedRowSize; //len; if (gotRow) {
pDataBlock->size += extendedRowSize; //len;
}
NEXT_TOKEN(pCxt->pSql, sToken); NEXT_TOKEN(pCxt->pSql, sToken);
if (TK_NK_RP != sToken.type) { if (TK_NK_RP != sToken.type) {
return buildSyntaxErrMsg(&pCxt->msg, ") expected", sToken.z); return buildSyntaxErrMsg(&pCxt->msg, ") expected", sToken.z);
} }
(*numOfRows)++; if (gotRow) {
(*numOfRows)++;
}
} }
if (0 == (*numOfRows)) { if (0 == (*numOfRows) && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL); return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1050,8 +1056,6 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
// for each table // for each table
while (1) { while (1) {
destroyInsertParseContextForTable(pCxt);
SToken sToken; SToken sToken;
char *tbName = NULL; char *tbName = NULL;
@ -1060,7 +1064,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
// no data in the sql string anymore. // no data in the sql string anymore.
if (sToken.n == 0) { if (sToken.n == 0) {
if (0 == pCxt->totalNum) { if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");; return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");;
} }
break; break;
@ -1070,6 +1074,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
return buildInvalidOperationMsg(&pCxt->msg, "single table allowed in one stmt");; return buildInvalidOperationMsg(&pCxt->msg, "single table allowed in one stmt");;
} }
destroyInsertParseContextForTable(pCxt);
if (TK_NK_QUESTION == sToken.type) { if (TK_NK_QUESTION == sToken.type) {
if (pCxt->pStmtCb) { if (pCxt->pStmtCb) {
CHECK_CODE((*pCxt->pStmtCb->getTbNameFn)(pCxt->pStmtCb->pStmt, &tbName)); CHECK_CODE((*pCxt->pStmtCb->getTbNameFn)(pCxt->pStmtCb->pStmt, &tbName));
@ -1105,7 +1111,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
if (TK_VALUES == sToken.type) { if (TK_VALUES == sToken.type) {
// pSql -> (field1_value, ...) [(field1_value2, ...) ...] // pSql -> (field1_value, ...) [(field1_value2, ...) ...]
CHECK_CODE(parseValuesClause(pCxt, dataBuf)); CHECK_CODE(parseValuesClause(pCxt, dataBuf));
pCxt->pOutput->insertType = TSDB_QUERY_TYPE_INSERT; TSDB_QUERY_SET_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_INSERT);
tbNum++; tbNum++;
continue; continue;

24
source/libs/parser/src/parInsertData.c
View File

@ -572,14 +572,20 @@ int initRowBuilder(SRowBuilder *pBuilder, int16_t schemaVer, SParsedDataColInfo
} }
void qResetStmtDataBlock(void* block, bool freeData) { int32_t qResetStmtDataBlock(void* block, bool keepBuf) {
STableDataBlocks* pBlock = (STableDataBlocks*)block; STableDataBlocks* pBlock = (STableDataBlocks*)block;
if (freeData) { if (keepBuf) {
taosMemoryFree(pBlock->pData); taosMemoryFreeClear(pBlock->pData);
pBlock->pData = taosMemoryMalloc(TSDB_PAYLOAD_SIZE);
if (NULL == pBlock->pData) {
return TSDB_CODE_OUT_OF_MEMORY;
}
memset(pBlock->pData, 0, sizeof(SSubmitBlk));
} else {
pBlock->pData = NULL;
} }
pBlock->pData = NULL;
pBlock->ordered = true; pBlock->ordered = true;
pBlock->prevTS = INT64_MIN; pBlock->prevTS = INT64_MIN;
pBlock->size = sizeof(SSubmitBlk); pBlock->size = sizeof(SSubmitBlk);
@ -589,6 +595,8 @@ void qResetStmtDataBlock(void* block, bool freeData) {
pBlock->headerSize = pBlock->size; pBlock->headerSize = pBlock->size;
memset(&pBlock->rowBuilder, 0, sizeof(pBlock->rowBuilder)); memset(&pBlock->rowBuilder, 0, sizeof(pBlock->rowBuilder));
return TSDB_CODE_SUCCESS;
} }
@ -601,9 +609,7 @@ int32_t qCloneStmtDataBlock(void** pDst, void* pSrc) {
memcpy(*pDst, pSrc, sizeof(STableDataBlocks)); memcpy(*pDst, pSrc, sizeof(STableDataBlocks));
((STableDataBlocks*)(*pDst))->cloned = true; ((STableDataBlocks*)(*pDst))->cloned = true;
qResetStmtDataBlock(*pDst, false); return qResetStmtDataBlock(*pDst, false);
return TSDB_CODE_SUCCESS;
} }
int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc) { int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc) {
@ -619,6 +625,8 @@ int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
memset(pBlock->pData, 0, sizeof(SSubmitBlk));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }

102
tests/script/api/batchprepare.c
View File

@ -19,6 +19,10 @@ void taosMsleep(int mseconds);
int32_t taosGetTimeOfDay(struct timeval *tv) { int32_t taosGetTimeOfDay(struct timeval *tv) {
return gettimeofday(tv, NULL); return gettimeofday(tv, NULL);
} }
void *taosMemoryMalloc(int32_t size) {
return malloc(size);
}
void *taosMemoryCalloc(int32_t num, int32_t size) { void *taosMemoryCalloc(int32_t num, int32_t size) {
return calloc(num, size); return calloc(num, size);
} }
@ -34,6 +38,51 @@ static int64_t taosGetTimestampUs() {
return (int64_t)systemTime.tv_sec * 1000000L + (int64_t)systemTime.tv_usec; return (int64_t)systemTime.tv_sec * 1000000L + (int64_t)systemTime.tv_usec;
} }
int stmt_allcol_func1(TAOS_STMT *stmt) {
struct {
int64_t ts;
int32_t v4;
} v = {0};
int32_t len[10] = {sizeof(v.ts), sizeof(v.v4)};
TAOS_BIND_v2 params[10];
params[0].buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
params[0].buffer_length = sizeof(v.ts);
params[0].buffer = &v.ts;
params[0].length = &len[0];
params[0].is_null = NULL;
params[0].num = 1;
params[1].buffer_type = TSDB_DATA_TYPE_TINYINT;
params[1].buffer_length = sizeof(v.v4);
params[1].buffer = &v.v4;
params[1].length = &len[1];
params[1].is_null = NULL;
params[1].num = 1;
char *sql = "insert into m0 values(?,?)";
int code = taos_stmt_prepare(stmt, sql, 0);
if (code != 0){
printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
}
v.ts = 1591060628000;
v.v4 = 111;
taos_stmt_bind_param(stmt, params);
taos_stmt_add_batch(stmt);
if (taos_stmt_execute(stmt) != 0) {
printf("failed to execute insert statement, error:%s.\n", taos_stmt_errstr(stmt));
exit(1);
}
return 0;
}
int stmt_scol_func1(TAOS_STMT *stmt) { int stmt_scol_func1(TAOS_STMT *stmt) {
struct { struct {
int64_t ts; int64_t ts;
@ -95,7 +144,7 @@ int stmt_scol_func1(TAOS_STMT *stmt) {
char *sql = "insert into ? (ts, v1,v2,f4,bin,bin2) values(?,?,?,?,?,?)"; char *sql = "insert into ? (ts, v1,v2,f4,bin,bin2) values(?,?,?,?,?,?)";
int code = taos_stmt_prepare(stmt, sql, 0); int code = taos_stmt_prepare(stmt, sql, 0);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_prepare. code:0x%x\n", code); printf("failed to execute taos_stmt_prepare. error:%s\n", taos_stmt_errstr(stmt));
} }
for (int zz = 0; zz < 10; zz++) { for (int zz = 0; zz < 10; zz++) {
@ -103,7 +152,7 @@ int stmt_scol_func1(TAOS_STMT *stmt) {
sprintf(buf, "m%d", zz); sprintf(buf, "m%d", zz);
code = taos_stmt_set_tbname(stmt, buf); code = taos_stmt_set_tbname(stmt, buf);
if (code != 0){ if (code != 0){
printf("failed to execute taos_stmt_set_tbname. code:0x%x\n", code); printf("failed to execute taos_stmt_set_tbname. error:%s\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
v.ts = 1591060628000 + zz * 10; v.ts = 1591060628000 + zz * 10;
@ -127,14 +176,13 @@ int stmt_scol_func1(TAOS_STMT *stmt) {
} }
if (taos_stmt_execute(stmt) != 0) { if (taos_stmt_execute(stmt) != 0) {
printf("failed to execute insert statement.\n"); printf("failed to execute insert statement, error:%s.\n", taos_stmt_errstr(stmt));
exit(1); exit(1);
} }
return 0; return 0;
} }
#if 0
int stmt_scol_func2(TAOS_STMT *stmt) { int stmt_scol_func2(TAOS_STMT *stmt) {
struct { struct {
@ -150,42 +198,48 @@ int stmt_scol_func2(TAOS_STMT *stmt) {
char blob[80]; char blob[80];
} v = {0}; } v = {0};
TAOS_BIND params[10]; TAOS_BIND_v2 params[10];
params[0].buffer_type = TSDB_DATA_TYPE_TIMESTAMP; params[0].buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
params[0].buffer_length = sizeof(v.ts); params[0].buffer_length = sizeof(v.ts);
params[0].buffer = &v.ts; params[0].buffer = &v.ts;
params[0].length = &params[0].buffer_length; params[0].length = &params[0].buffer_length;
params[0].is_null = NULL; params[0].is_null = NULL;
params[0].num = 1;
params[1].buffer_type = TSDB_DATA_TYPE_TINYINT; params[1].buffer_type = TSDB_DATA_TYPE_TINYINT;
params[1].buffer_length = sizeof(v.v1); params[1].buffer_length = sizeof(v.v1);
params[1].buffer = &v.v1; params[1].buffer = &v.v1;
params[1].length = &params[1].buffer_length; params[1].length = &params[1].buffer_length;
params[1].is_null = NULL; params[1].is_null = NULL;
params[1].num = 1;
params[2].buffer_type = TSDB_DATA_TYPE_SMALLINT; params[2].buffer_type = TSDB_DATA_TYPE_SMALLINT;
params[2].buffer_length = sizeof(v.v2); params[2].buffer_length = sizeof(v.v2);
params[2].buffer = &v.v2; params[2].buffer = &v.v2;
params[2].length = &params[2].buffer_length; params[2].length = &params[2].buffer_length;
params[2].is_null = NULL; params[2].is_null = NULL;
params[2].num = 1;
params[3].buffer_type = TSDB_DATA_TYPE_FLOAT; params[3].buffer_type = TSDB_DATA_TYPE_FLOAT;
params[3].buffer_length = sizeof(v.f4); params[3].buffer_length = sizeof(v.f4);
params[3].buffer = &v.f4; params[3].buffer = &v.f4;
params[3].length = &params[3].buffer_length; params[3].length = &params[3].buffer_length;
params[3].is_null = NULL; params[3].is_null = NULL;
params[3].num = 1;
params[4].buffer_type = TSDB_DATA_TYPE_BINARY; params[4].buffer_type = TSDB_DATA_TYPE_BINARY;
params[4].buffer_length = sizeof(v.bin); params[4].buffer_length = sizeof(v.bin);
params[4].buffer = v.bin; params[4].buffer = v.bin;
params[4].length = &params[4].buffer_length; params[4].length = &params[4].buffer_length;
params[4].is_null = NULL; params[4].is_null = NULL;
params[4].num = 1;
params[5].buffer_type = TSDB_DATA_TYPE_BINARY; params[5].buffer_type = TSDB_DATA_TYPE_BINARY;
params[5].buffer_length = sizeof(v.bin); params[5].buffer_length = sizeof(v.bin);
params[5].buffer = v.bin; params[5].buffer = v.bin;
params[5].length = &params[5].buffer_length; params[5].length = &params[5].buffer_length;
params[5].is_null = NULL; params[5].is_null = NULL;
params[5].num = 1;
char *sql = "insert into m0 (ts, v1,v2,f4,bin,bin2) values(?,?,?,?,?,?)"; char *sql = "insert into m0 (ts, v1,v2,f4,bin,bin2) values(?,?,?,?,?,?)";
int code = taos_stmt_prepare(stmt, sql, 0); int code = taos_stmt_prepare(stmt, sql, 0);
@ -243,7 +297,7 @@ int stmt_scol_func3(TAOS_STMT *stmt) {
int *lb = taosMemoryMalloc(60 * sizeof(int)); int *lb = taosMemoryMalloc(60 * sizeof(int));
TAOS_MULTI_BIND *params = taosMemoryCalloc(1, sizeof(TAOS_MULTI_BIND) * 900000*10); TAOS_BIND_v2 *params = taosMemoryCalloc(1, sizeof(TAOS_BIND_v2) * 900000*10);
char* is_null = taosMemoryMalloc(sizeof(char) * 60); char* is_null = taosMemoryMalloc(sizeof(char) * 60);
char* no_null = taosMemoryMalloc(sizeof(char) * 60); char* no_null = taosMemoryMalloc(sizeof(char) * 60);
@ -311,7 +365,7 @@ int stmt_scol_func3(TAOS_STMT *stmt) {
v.ts[i] = tts + i; v.ts[i] = tts + i;
} }
unsigned long long starttime = getCurrentTime(); unsigned long long starttime = taosGetTimestampUs();
char *sql = "insert into ? (ts, v1,v2,f4,bin,bin2) values(?,?,?,?,?,?)"; char *sql = "insert into ? (ts, v1,v2,f4,bin,bin2) values(?,?,?,?,?,?)";
int code = taos_stmt_prepare(stmt, sql, 0); int code = taos_stmt_prepare(stmt, sql, 0);
@ -341,7 +395,7 @@ int stmt_scol_func3(TAOS_STMT *stmt) {
++id; ++id;
} }
unsigned long long endtime = getCurrentTime(); unsigned long long endtime = taosGetTimestampUs();
printf("insert total %d records, used %u seconds, avg:%u useconds\n", 3000*300*60, (endtime-starttime)/1000000UL, (endtime-starttime)/(3000*300*60)); printf("insert total %d records, used %u seconds, avg:%u useconds\n", 3000*300*60, (endtime-starttime)/1000000UL, (endtime-starttime)/(3000*300*60));
taosMemoryFree(v.ts); taosMemoryFree(v.ts);
@ -353,6 +407,7 @@ int stmt_scol_func3(TAOS_STMT *stmt) {
return 0; return 0;
} }
#if 0
//10 tables 10 records single column bind //10 tables 10 records single column bind
@ -4516,11 +4571,33 @@ void* runcase(void *par) {
(void)idx; (void)idx;
#if 0
prepare(taos, 0, 1);
stmt = taos_stmt_init(taos);
if (NULL == stmt) {
printf("taos_stmt_init failed\n");
exit(1);
}
printf("1t+1records start\n");
stmt_allcol_func1(stmt);
printf("1t+1records end\n");
printf("check result start\n");
check_result(taos, "m0", 1, 1);
printf("check result end\n");
taos_stmt_close(stmt);
#endif
#if 1
#if 0
prepare(taos, 1, 1); prepare(taos, 1, 1);
stmt = taos_stmt_init(taos); stmt = taos_stmt_init(taos);
if (NULL == stmt) {
printf("taos_stmt_init failed\n");
exit(1);
}
printf("10t+10records+specifycol start\n"); printf("10t+10records+specifycol start\n");
stmt_scol_func1(stmt); stmt_scol_func1(stmt);
@ -4540,8 +4617,10 @@ void* runcase(void *par) {
taos_stmt_close(stmt); taos_stmt_close(stmt);
#endif #endif
#if 0 #if 0
#if 1
prepare(taos, 1, 1); prepare(taos, 1, 1);
stmt = taos_stmt_init(taos); stmt = taos_stmt_init(taos);
@ -4556,6 +4635,7 @@ void* runcase(void *par) {
#endif #endif
#if 1 #if 1
prepare(taos, 1, 1); prepare(taos, 1, 1);
@ -4575,6 +4655,8 @@ void* runcase(void *par) {
#endif #endif
#if 0
#if 1 #if 1
prepare(taos, 1, 1); prepare(taos, 1, 1);

6
tests/script/api/makefile
View File

@ -6,16 +6,12 @@ TARGET=exe
LFLAGS = '-Wl,-rpath,/usr/local/taos/driver/' -ltaos -lpthread -lm -lrt LFLAGS = '-Wl,-rpath,/usr/local/taos/driver/' -ltaos -lpthread -lm -lrt
CFLAGS = -O0 -g -Wall -Wno-deprecated -fPIC -Wno-unused-result -Wconversion \ CFLAGS = -O0 -g -Wall -Wno-deprecated -fPIC -Wno-unused-result -Wconversion \
-Wno-char-subscripts -D_REENTRANT -Wno-format -D_REENTRANT -DLINUX \ -Wno-char-subscripts -D_REENTRANT -Wno-format -D_REENTRANT -DLINUX \
-Wno-unused-function -D_M_X64 -I/usr/local/taos/include -std=gnu99 \ -Wno-unused-function -D_M_X64 -I/usr/local/taos/include -std=gnu99
-fsanitize=address
all: $(TARGET) all: $(TARGET)
exe: exe:
gcc $(CFLAGS) ./batchprepare.c -o $(ROOT)batchprepare $(LFLAGS) gcc $(CFLAGS) ./batchprepare.c -o $(ROOT)batchprepare $(LFLAGS)
gcc $(CFLAGS) ./stmtBatchTest.c -o $(ROOT)stmtBatchTest $(LFLAGS)
gcc $(CFLAGS) ./stmtTest.c -o $(ROOT)stmtTest $(LFLAGS)
gcc $(CFLAGS) ./stmt_function.c -o $(ROOT)stmt_function $(LFLAGS)
clean: clean:
rm $(ROOT)batchprepare rm $(ROOT)batchprepare