diff --git a/include/common/tglobal.h b/include/common/tglobal.h index bd5e74387edc4292e3a019b817a8d70f266e081d..350cd785d96c838345366453fae5ba935d6dd9eb 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -103,6 +103,7 @@ extern bool tsKeepColumnName; // client extern int32_t tsMinSlidingTime; extern int32_t tsMinIntervalTime; +extern int32_t tsMaxMemUsedByInsert; // build info extern char version[]; diff --git a/include/libs/catalog/catalog.h b/include/libs/catalog/catalog.h index ee566d759a980cf42cc8a77e81a7cc2e639b707c..c3caac00ad62a030babfad2843e7e3eb47884dc3 100644 --- a/include/libs/catalog/catalog.h +++ b/include/libs/catalog/catalog.h @@ -303,6 +303,10 @@ int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t re int32_t catalogClearCache(void); +SMetaData* catalogCloneMetaData(SMetaData* pData); + +void catalogFreeMetaData(SMetaData* pData); + /** * Destroy catalog and relase all resources */ diff --git a/include/libs/nodes/querynodes.h b/include/libs/nodes/querynodes.h index 5ee097bd9254970536c61caace0fdde3f6a11d9d..e1acf0dd6ac74954c3be320e9792b7a78206a4bf 100644 --- a/include/libs/nodes/querynodes.h +++ b/include/libs/nodes/querynodes.h @@ -385,7 +385,6 @@ typedef struct SCmdMsgInfo { SEpSet epSet; void* pMsg; int32_t msgLen; - void* pExtension; // todo remove it soon } SCmdMsgInfo; typedef enum EQueryExecMode { diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index b1a937910dfe8defd107ec525afd20edfc639aaf..bcd2316baf22f896916bb0eb5b3f6ae654afb48a 100644 --- a/include/libs/parser/parser.h +++ b/include/libs/parser/parser.h @@ -33,6 +33,13 @@ typedef struct SStmtCallback { int32_t (*getExecInfoFn)(TAOS_STMT*, SHashObj**, SHashObj**); } SStmtCallback; +typedef struct SParseCsvCxt { + TdFilePtr fp; // last parsed file + int32_t tableNo; // last parsed table + SName tableName; // last parsed table + const char* pLastSqlPos; // the location of the last parsed sql +} SParseCsvCxt; + typedef struct SParseContext { uint64_t requestId; int64_t requestRid; @@ -57,6 +64,8 @@ typedef struct SParseContext { SArray* pTableMetaPos; // sql table pos => catalog data pos SArray* pTableVgroupPos; // sql table pos => catalog data pos int64_t allocatorId; + bool needMultiParse; + SParseCsvCxt csvCxt; } SParseContext; int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery); @@ -67,6 +76,8 @@ int32_t qParseSqlSyntax(SParseContext* pCxt, SQuery** pQuery, struct SCatalogReq int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCatalogReq, const struct SMetaData* pMetaData, SQuery* pQuery); +void qDestroyParseContext(SParseContext* pCxt); + void qDestroyQuery(SQuery* pQueryNode); int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema); diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index 535a436c6c17dfd90603f40dc189bcff5b2902d2..f815adfeaa8e77473acd67a01f945aa360e1d5e7 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -380,16 +380,25 @@ void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey); // --- mq void hbMgrInitMqHbRspHandle(); +typedef struct SSqlCallbackWrapper { + SParseContext* pParseCtx; + SCatalogReq* pCatalogReq; + SMetaData* pResultMeta; + SRequestObj* pRequest; +} SSqlCallbackWrapper; + SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res); int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList); -void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta); -int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest); -int32_t updateQnodeList(SAppInstInfo* pInfo, SArray* pNodeList); -void doAsyncQuery(SRequestObj* pRequest, bool forceUpdateMeta); -int32_t removeMeta(STscObj* pTscObj, SArray* tbList); -int32_t handleAlterTbExecRes(void* res, struct SCatalog* pCatalog); -int32_t handleCreateTbExecRes(void* res, SCatalog* pCatalog); -bool qnodeRequired(SRequestObj* pRequest); +void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, SSqlCallbackWrapper* pWrapper); +int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest); +int32_t updateQnodeList(SAppInstInfo* pInfo, SArray* pNodeList); +void doAsyncQuery(SRequestObj* pRequest, bool forceUpdateMeta); +int32_t removeMeta(STscObj* pTscObj, SArray* tbList); +int32_t handleAlterTbExecRes(void* res, struct SCatalog* pCatalog); +int32_t handleCreateTbExecRes(void* res, SCatalog* pCatalog); +bool qnodeRequired(SRequestObj* pRequest); +int32_t continueInsertFromCsv(SSqlCallbackWrapper* pWrapper, SRequestObj* pRequest); +void destorySqlCallbackWrapper(SSqlCallbackWrapper* pWrapper); #ifdef __cplusplus } diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 8ffc88ec28fea8de7059d97fface02ad42c0eadf..08e844dc2c3e4c88bc46fdd35c4e0a7e9f445dc8 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -868,10 +868,11 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) { return code; } -//todo refacto the error code mgmt +// todo refacto the error code mgmt void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) { - SRequestObj* pRequest = (SRequestObj*)param; - STscObj* pTscObj = pRequest->pTscObj; + SSqlCallbackWrapper* pWrapper = param; + SRequestObj* pRequest = pWrapper->pRequest; + STscObj* pTscObj = pRequest->pTscObj; pRequest->code = code; if (pResult) { @@ -882,7 +883,7 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) { int32_t type = pRequest->type; if (TDMT_VND_SUBMIT == type || TDMT_VND_DELETE == type || TDMT_VND_CREATE_TABLE == type) { if (pResult) { - pRequest->body.resInfo.numOfRows = pResult->numOfRows; + pRequest->body.resInfo.numOfRows += pResult->numOfRows; // record the insert rows if (TDMT_VND_SUBMIT == type) { @@ -899,12 +900,13 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) { pRequest->requestId); if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code) && pRequest->sqlstr != NULL) { - tscDebug("0x%" PRIx64 " client retry to handle the error, code:%s, tryCount:%d, reqId:0x%" PRIx64, - pRequest->self, tstrerror(code), pRequest->retry, pRequest->requestId); + tscDebug("0x%" PRIx64 " client retry to handle the error, code:%s, tryCount:%d, reqId:0x%" PRIx64, pRequest->self, + tstrerror(code), pRequest->retry, pRequest->requestId); pRequest->prevCode = code; schedulerFreeJob(&pRequest->body.queryJob, 0); qDestroyQuery(pRequest->pQuery); pRequest->pQuery = NULL; + destorySqlCallbackWrapper(pWrapper); doAsyncQuery(pRequest, true); return; } @@ -920,6 +922,15 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) { pRequest->code = code1; } + if (pRequest->code == TSDB_CODE_SUCCESS && pWrapper->pParseCtx->needMultiParse) { + code = continueInsertFromCsv(pWrapper, pRequest); + if (TSDB_CODE_SUCCESS == code) { + return; + } + } + + destorySqlCallbackWrapper(pWrapper); + // return to client pRequest->body.queryFp(pRequest->body.param, pRequest, code); } @@ -1020,76 +1031,86 @@ SRequestObj* launchQuery(uint64_t connId, const char* sql, int sqlLen, bool vali return launchQueryImpl(pRequest, pQuery, false, NULL); } -void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta) { +static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, + SSqlCallbackWrapper* pWrapper) { + pRequest->type = pQuery->msgType; + + SArray* pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad)); + SPlanContext cxt = {.queryId = pRequest->requestId, + .acctId = pRequest->pTscObj->acctId, + .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp), + .pAstRoot = pQuery->pRoot, + .showRewrite = pQuery->showRewrite, + .pMsg = pRequest->msgBuf, + .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE, + .pUser = pRequest->pTscObj->user, + .sysInfo = pRequest->pTscObj->sysInfo, + .allocatorId = pRequest->allocatorRefId}; + SQueryPlan* pDag = NULL; + int32_t code = qCreateQueryPlan(&cxt, &pDag, pMnodeList); + if (code) { + tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code), + pRequest->requestId); + } else { + pRequest->body.subplanNum = pDag->numOfSubplans; + } + + pRequest->metric.planEnd = taosGetTimestampUs(); + + if (TSDB_CODE_SUCCESS == code && !pRequest->validateOnly) { + SArray* pNodeList = NULL; + buildAsyncExecNodeList(pRequest, &pNodeList, pMnodeList, pResultMeta); + + SRequestConnInfo conn = {.pTrans = getAppInfo(pRequest)->pTransporter, + .requestId = pRequest->requestId, + .requestObjRefId = pRequest->self}; + SSchedulerReq req = { + .syncReq = false, + .localReq = (tsQueryPolicy == QUERY_POLICY_CLIENT), + .pConn = &conn, + .pNodeList = pNodeList, + .pDag = pDag, + .allocatorRefId = pRequest->allocatorRefId, + .sql = pRequest->sqlstr, + .startTs = pRequest->metric.start, + .execFp = schedulerExecCb, + .cbParam = pWrapper, + .chkKillFp = chkRequestKilled, + .chkKillParam = (void*)pRequest->self, + .pExecRes = NULL, + }; + code = schedulerExecJob(&req, &pRequest->body.queryJob); + taosArrayDestroy(pNodeList); + } else { + tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code), + pRequest->requestId); + destorySqlCallbackWrapper(pWrapper); + pRequest->body.queryFp(pRequest->body.param, pRequest, code); + } + + // todo not to be released here + taosArrayDestroy(pMnodeList); + + return code; +} + +void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, SSqlCallbackWrapper* pWrapper) { int32_t code = 0; pRequest->body.execMode = pQuery->execMode; + if (QUERY_EXEC_MODE_SCHEDULE != pRequest->body.execMode) { + destorySqlCallbackWrapper(pWrapper); + } switch (pQuery->execMode) { case QUERY_EXEC_MODE_LOCAL: asyncExecLocalCmd(pRequest, pQuery); - return; + break; case QUERY_EXEC_MODE_RPC: code = asyncExecDdlQuery(pRequest, pQuery); break; case QUERY_EXEC_MODE_SCHEDULE: { - SArray* pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad)); - - pRequest->type = pQuery->msgType; - - SPlanContext cxt = {.queryId = pRequest->requestId, - .acctId = pRequest->pTscObj->acctId, - .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp), - .pAstRoot = pQuery->pRoot, - .showRewrite = pQuery->showRewrite, - .pMsg = pRequest->msgBuf, - .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE, - .pUser = pRequest->pTscObj->user, - .sysInfo = pRequest->pTscObj->sysInfo, - .allocatorId = pRequest->allocatorRefId}; - - SAppInstInfo* pAppInfo = getAppInfo(pRequest); - SQueryPlan* pDag = NULL; - code = qCreateQueryPlan(&cxt, &pDag, pMnodeList); - if (code) { - tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code), - pRequest->requestId); - } else { - pRequest->body.subplanNum = pDag->numOfSubplans; - } - - pRequest->metric.planEnd = taosGetTimestampUs(); - if (TSDB_CODE_SUCCESS == code && !pRequest->validateOnly) { - SArray* pNodeList = NULL; - buildAsyncExecNodeList(pRequest, &pNodeList, pMnodeList, pResultMeta); - - SRequestConnInfo conn = { - .pTrans = pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self}; - SSchedulerReq req = { - .syncReq = false, - .localReq = (tsQueryPolicy == QUERY_POLICY_CLIENT), - .pConn = &conn, - .pNodeList = pNodeList, - .pDag = pDag, - .allocatorRefId = pRequest->allocatorRefId, - .sql = pRequest->sqlstr, - .startTs = pRequest->metric.start, - .execFp = schedulerExecCb, - .cbParam = pRequest, - .chkKillFp = chkRequestKilled, - .chkKillParam = (void*)pRequest->self, - .pExecRes = NULL, - }; - code = schedulerExecJob(&req, &pRequest->body.queryJob); - taosArrayDestroy(pNodeList); - } else { - tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code), - pRequest->requestId); - pRequest->body.queryFp(pRequest->body.param, pRequest, code); - } - - // todo not to be released here - taosArrayDestroy(pMnodeList); + code = asyncExecSchQuery(pRequest, pQuery, pResultMeta, pWrapper); break; } case QUERY_EXEC_MODE_EMPTY_RESULT: diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c index 5b255a28ea8e125145f1bdb2a2bfcca5d6fc61f5..6f8cef7c0d50aff6dd76903691a0fc3521230d85 100644 --- a/source/client/src/clientMain.c +++ b/source/client/src/clientMain.c @@ -667,43 +667,47 @@ const char *taos_get_server_info(TAOS *taos) { return pTscObj->sDetailVer; } -typedef struct SqlParseWrapper { - SParseContext *pCtx; - SCatalogReq catalogReq; - SRequestObj *pRequest; -} SqlParseWrapper; - static void destoryTablesReq(void *p) { STablesReq *pRes = (STablesReq *)p; taosArrayDestroy(pRes->pTables); } -static void destorySqlParseWrapper(SqlParseWrapper *pWrapper) { - taosArrayDestroy(pWrapper->catalogReq.pDbVgroup); - taosArrayDestroy(pWrapper->catalogReq.pDbCfg); - taosArrayDestroy(pWrapper->catalogReq.pDbInfo); - taosArrayDestroyEx(pWrapper->catalogReq.pTableMeta, destoryTablesReq); - taosArrayDestroyEx(pWrapper->catalogReq.pTableHash, destoryTablesReq); - taosArrayDestroy(pWrapper->catalogReq.pUdf); - taosArrayDestroy(pWrapper->catalogReq.pIndex); - taosArrayDestroy(pWrapper->catalogReq.pUser); - taosArrayDestroy(pWrapper->catalogReq.pTableIndex); - taosArrayDestroy(pWrapper->pCtx->pTableMetaPos); - taosArrayDestroy(pWrapper->pCtx->pTableVgroupPos); - taosMemoryFree(pWrapper->pCtx); +static void destoryCatalogReq(SCatalogReq *pCatalogReq) { + if (NULL == pCatalogReq) { + return; + } + taosArrayDestroy(pCatalogReq->pDbVgroup); + taosArrayDestroy(pCatalogReq->pDbCfg); + taosArrayDestroy(pCatalogReq->pDbInfo); + taosArrayDestroyEx(pCatalogReq->pTableMeta, destoryTablesReq); + taosArrayDestroyEx(pCatalogReq->pTableHash, destoryTablesReq); + taosArrayDestroy(pCatalogReq->pUdf); + taosArrayDestroy(pCatalogReq->pIndex); + taosArrayDestroy(pCatalogReq->pUser); + taosArrayDestroy(pCatalogReq->pTableIndex); + taosMemoryFree(pCatalogReq); +} + +void destorySqlCallbackWrapper(SSqlCallbackWrapper *pWrapper) { + if (NULL == pWrapper) { + return; + } + destoryCatalogReq(pWrapper->pCatalogReq); + qDestroyParseContext(pWrapper->pParseCtx); + catalogFreeMetaData(pWrapper->pResultMeta); taosMemoryFree(pWrapper); } void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) { - SqlParseWrapper *pWrapper = (SqlParseWrapper *)param; - SRequestObj *pRequest = pWrapper->pRequest; - SQuery *pQuery = pRequest->pQuery; + SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)param; + SRequestObj *pRequest = pWrapper->pRequest; + SQuery *pQuery = pRequest->pQuery; pRequest->metric.ctgEnd = taosGetTimestampUs(); qDebug("0x%" PRIx64 " start to semantic analysis, reqId:0x%" PRIx64, pRequest->self, pRequest->requestId); if (code == TSDB_CODE_SUCCESS) { - code = qAnalyseSqlSemantic(pWrapper->pCtx, &pWrapper->catalogReq, pResultMeta, pQuery); + code = qAnalyseSqlSemantic(pWrapper->pParseCtx, pWrapper->pCatalogReq, pResultMeta, pQuery); pRequest->stableQuery = pQuery->stableQuery; if (pQuery->pRoot) { pRequest->stmtType = pQuery->pRoot->type; @@ -712,6 +716,13 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) { pRequest->metric.semanticEnd = taosGetTimestampUs(); + if (code == TSDB_CODE_SUCCESS && pWrapper->pParseCtx->needMultiParse) { + pWrapper->pResultMeta = catalogCloneMetaData(pResultMeta); + if (NULL == pWrapper->pResultMeta) { + code = TSDB_CODE_OUT_OF_MEMORY; + } + } + if (code == TSDB_CODE_SUCCESS) { if (pQuery->haveResultSet) { setResSchemaInfo(&pRequest->body.resInfo, pQuery->pResSchema, pQuery->numOfResCols); @@ -722,15 +733,13 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) { TSWAP(pRequest->tableList, (pQuery)->pTableList); TSWAP(pRequest->targetTableList, (pQuery)->pTargetTableList); - destorySqlParseWrapper(pWrapper); - - double el = (pRequest->metric.semanticEnd - pRequest->metric.ctgEnd)/1000.0; + double el = (pRequest->metric.semanticEnd - pRequest->metric.ctgEnd) / 1000.0; tscDebug("0x%" PRIx64 " analysis semantics completed, start async query, elapsed time:%.2f ms, reqId:0x%" PRIx64, pRequest->self, el, pRequest->requestId); - launchAsyncQuery(pRequest, pQuery, pResultMeta); + launchAsyncQuery(pRequest, pQuery, pResultMeta, pWrapper); } else { - destorySqlParseWrapper(pWrapper); + destorySqlCallbackWrapper(pWrapper); qDestroyQuery(pRequest->pQuery); pRequest->pQuery = NULL; @@ -750,6 +759,16 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) { } } +int32_t continueInsertFromCsv(SSqlCallbackWrapper *pWrapper, SRequestObj *pRequest) { + qDestroyQuery(pRequest->pQuery); + pRequest->pQuery = (SQuery *)nodesMakeNode(QUERY_NODE_QUERY); + if (NULL == pRequest->pQuery) { + return TSDB_CODE_OUT_OF_MEMORY; + } + retrieveMetaCallback(pWrapper->pResultMeta, pWrapper, TSDB_CODE_SUCCESS); + return TSDB_CODE_SUCCESS; +} + void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) { int64_t connId = *(int64_t *)taos; taosAsyncQueryImpl(connId, sql, fp, param, false); @@ -786,37 +805,48 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) { } void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { - SParseContext *pCxt = NULL; - STscObj *pTscObj = pRequest->pTscObj; - int32_t code = 0; + STscObj *pTscObj = pRequest->pTscObj; + SSqlCallbackWrapper *pWrapper = NULL; + int32_t code = TSDB_CODE_SUCCESS; if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) { code = pRequest->prevCode; - goto _error; } - code = createParseContext(pRequest, &pCxt); - if (code != TSDB_CODE_SUCCESS) { - goto _error; + if (TSDB_CODE_SUCCESS == code) { + pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper)); + if (pWrapper == NULL) { + code = TSDB_CODE_OUT_OF_MEMORY; + } else { + pWrapper->pRequest = pRequest; + } } - pCxt->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp); - code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCxt->pCatalog); - if (code != TSDB_CODE_SUCCESS) { - goto _error; + if (TSDB_CODE_SUCCESS == code) { + code = createParseContext(pRequest, &pWrapper->pParseCtx); } - pRequest->metric.syntaxStart = taosGetTimestampUs(); - - SCatalogReq catalogReq = {.forceUpdate = updateMetaForce, .qNodeRequired = qnodeRequired(pRequest)}; - code = qParseSqlSyntax(pCxt, &pRequest->pQuery, &catalogReq); - if (code != TSDB_CODE_SUCCESS) { - goto _error; + if (TSDB_CODE_SUCCESS == code) { + pWrapper->pParseCtx->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp); + code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pWrapper->pParseCtx->pCatalog); } - pRequest->metric.syntaxEnd = taosGetTimestampUs(); + if (TSDB_CODE_SUCCESS == code) { + pRequest->metric.syntaxStart = taosGetTimestampUs(); + + pWrapper->pCatalogReq = taosMemoryCalloc(1, sizeof(SCatalogReq)); + if (pWrapper->pCatalogReq == NULL) { + code = TSDB_CODE_OUT_OF_MEMORY; + } else { + pWrapper->pCatalogReq->forceUpdate = updateMetaForce; + pWrapper->pCatalogReq->qNodeRequired = qnodeRequired(pRequest); + code = qParseSqlSyntax(pWrapper->pParseCtx, &pRequest->pQuery, pWrapper->pCatalogReq); + } + + pRequest->metric.syntaxEnd = taosGetTimestampUs(); + } - if (!updateMetaForce) { + if (TSDB_CODE_SUCCESS == code && !updateMetaForce) { SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary; if (NULL == pRequest->pQuery->pRoot) { atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1); @@ -825,38 +855,26 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) { } } - SqlParseWrapper *pWrapper = taosMemoryCalloc(1, sizeof(SqlParseWrapper)); - if (pWrapper == NULL) { - code = TSDB_CODE_OUT_OF_MEMORY; - goto _error; - } - - pWrapper->pCtx = pCxt; - pWrapper->pRequest = pRequest; - pWrapper->catalogReq = catalogReq; + if (TSDB_CODE_SUCCESS == code) { + SRequestConnInfo conn = {.pTrans = pWrapper->pParseCtx->pTransporter, + .requestId = pWrapper->pParseCtx->requestId, + .requestObjRefId = pWrapper->pParseCtx->requestRid, + .mgmtEps = pWrapper->pParseCtx->mgmtEpSet}; - SRequestConnInfo conn = {.pTrans = pCxt->pTransporter, - .requestId = pCxt->requestId, - .requestObjRefId = pCxt->requestRid, - .mgmtEps = pCxt->mgmtEpSet}; + pRequest->metric.ctgStart = taosGetTimestampUs(); - pRequest->metric.ctgStart = taosGetTimestampUs(); - - code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, &catalogReq, retrieveMetaCallback, pWrapper, - &pRequest->body.queryJob); - pCxt = NULL; - if (code == TSDB_CODE_SUCCESS) { - return; + code = catalogAsyncGetAllMeta(pWrapper->pParseCtx->pCatalog, &conn, pWrapper->pCatalogReq, retrieveMetaCallback, + pWrapper, &pRequest->body.queryJob); } -_error: - tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code), - pRequest->requestId); - taosMemoryFree(pCxt); - - terrno = code; - pRequest->code = code; - pRequest->body.queryFp(pRequest->body.param, pRequest, code); + if (TSDB_CODE_SUCCESS != code) { + tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code), + pRequest->requestId); + destorySqlCallbackWrapper(pWrapper); + terrno = code; + pRequest->code = code; + pRequest->body.queryFp(pRequest->body.param, pRequest, code); + } } static void fetchCallback(void *pResult, void *param, int32_t code) { diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index 195466061d2c19cbd6cedf6fadc08c8f2cc92bdc..4eb1a3712c0273ac4d9148765bc2bd89fd40fbb9 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -2336,7 +2336,7 @@ static int32_t smlInsertData(SSmlHandle *info) { SAppClusterSummary *pActivity = &info->taos->pAppInfo->summary; atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1); - launchAsyncQuery(info->pRequest, info->pQuery, NULL); + launchAsyncQuery(info->pRequest, info->pQuery, NULL, NULL); return TSDB_CODE_SUCCESS; } diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index ce9a9a7b5099825b731542fc71810b2d4bf867fd..88df54a9eace1c310af0c0e9787d8a8774803e97 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -85,7 +85,8 @@ uint16_t tsTelemPort = 80; char tsSmlTagName[TSDB_COL_NAME_LEN] = "_tag_null"; char tsSmlChildTableName[TSDB_TABLE_NAME_LEN] = ""; // user defined child table name can be specified in tag value. // If set to empty system will generate table name using MD5 hash. -bool tsSmlDataFormat = false; // true means that the name and order of cols in each line are the same(only for influx protocol) +// true means that the name and order of cols in each line are the same(only for influx protocol) +bool tsSmlDataFormat = false; // query int32_t tsQueryPolicy = 1; @@ -125,6 +126,9 @@ int32_t tsMaxNumOfDistinctResults = 1000 * 10000; // 1 database precision unit for interval time range, changed accordingly int32_t tsMinIntervalTime = 1; +// maximum memory allowed to be allocated for a single csv load (in MB) +int32_t tsMaxMemUsedByInsert = 1024; + // the maximum allowed query buffer size during query processing for each data node. // -1 no limit (default) // 0 no query allowed, queries are disabled @@ -296,6 +300,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) { if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1; if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1; if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1; + if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1; tsNumOfTaskQueueThreads = tsNumOfCores / 2; tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4); @@ -374,8 +379,8 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { tsNumOfVnodeStreamThreads = TMAX(tsNumOfVnodeStreamThreads, 4); if (cfgAddInt32(pCfg, "numOfVnodeStreamThreads", tsNumOfVnodeStreamThreads, 4, 1024, 0) != 0) return -1; -// tsNumOfVnodeFetchThreads = 1; -// if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 1, 1, 0) != 0) return -1; + // tsNumOfVnodeFetchThreads = 1; + // if (cfgAddInt32(pCfg, "numOfVnodeFetchThreads", tsNumOfVnodeFetchThreads, 1, 1, 0) != 0) return -1; tsNumOfVnodeWriteThreads = tsNumOfCores; tsNumOfVnodeWriteThreads = TMAX(tsNumOfVnodeWriteThreads, 1); @@ -497,15 +502,15 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) { pItem->stype = stype; } -/* - pItem = cfgGetItem(tsCfg, "numOfVnodeFetchThreads"); - if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { - tsNumOfVnodeFetchThreads = numOfCores / 4; - tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4); - pItem->i32 = tsNumOfVnodeFetchThreads; - pItem->stype = stype; - } -*/ + /* + pItem = cfgGetItem(tsCfg, "numOfVnodeFetchThreads"); + if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { + tsNumOfVnodeFetchThreads = numOfCores / 4; + tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4); + pItem->i32 = tsNumOfVnodeFetchThreads; + pItem->stype = stype; + } + */ pItem = cfgGetItem(tsCfg, "numOfVnodeWriteThreads"); if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { @@ -648,6 +653,8 @@ static int32_t taosSetClientCfg(SConfig *pCfg) { tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN); tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval; + tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32; + tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32; tsCompressMsgSize = cfgGetItem(pCfg, "compressMsgSize")->i32; tsCompressColData = cfgGetItem(pCfg, "compressColData")->i32; @@ -705,7 +712,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32; tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32; tsNumOfVnodeStreamThreads = cfgGetItem(pCfg, "numOfVnodeStreamThreads")->i32; -// tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; + // tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32; tsNumOfVnodeSyncThreads = cfgGetItem(pCfg, "numOfVnodeSyncThreads")->i32; tsNumOfVnodeRsmaThreads = cfgGetItem(pCfg, "numOfVnodeRsmaThreads")->i32; @@ -877,6 +884,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { tsMaxShellConns = cfgGetItem(pCfg, "maxShellConns")->i32; } else if (strcasecmp("maxNumOfDistinctRes", name) == 0) { tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32; + } else if (strcasecmp("maxMemUsedByInsert", name) == 0) { + tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32; } break; } @@ -955,10 +964,10 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32; } else if (strcasecmp("numOfVnodeQueryThreads", name) == 0) { tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32; -/* - } else if (strcasecmp("numOfVnodeFetchThreads", name) == 0) { - tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; -*/ + /* + } else if (strcasecmp("numOfVnodeFetchThreads", name) == 0) { + tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; + */ } else if (strcasecmp("numOfVnodeWriteThreads", name) == 0) { tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32; } else if (strcasecmp("numOfVnodeSyncThreads", name) == 0) { diff --git a/source/libs/catalog/src/ctgUtil.c b/source/libs/catalog/src/ctgUtil.c index 97b174de1c7c70553011aae62b9dcf3e779c0efb..0c0ca9649a5962e0b1c4bd49796c25f7bc52912a 100644 --- a/source/libs/catalog/src/ctgUtil.c +++ b/source/libs/catalog/src/ctgUtil.c @@ -1193,4 +1193,232 @@ SName* ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch) { return (SName*)taosArrayGet(pReq->pTables, pFetch->tbIdx); } +static void* ctgCloneDbVgroup(void* pSrc) { + return taosArrayDup((const SArray*)pSrc); +} + +static void ctgFreeDbVgroup(void* p) { + taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); +} + +static void* ctgCloneDbCfgInfo(void* pSrc) { + SDbCfgInfo* pDst = taosMemoryMalloc(sizeof(SDbCfgInfo)); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, sizeof(SDbCfgInfo)); + return pDst; +} + +static void ctgFreeDbCfgInfo(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneDbInfo(void* pSrc) { + SDbInfo* pDst = taosMemoryMalloc(sizeof(SDbInfo)); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, sizeof(SDbInfo)); + return pDst; +} + +static void ctgFreeDbInfo(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneTableMeta(void* pSrc) { + STableMeta* pMeta = pSrc; + int32_t size = sizeof(STableMeta) + (pMeta->tableInfo.numOfColumns + pMeta->tableInfo.numOfTags) * sizeof(SSchema); + STableMeta* pDst = taosMemoryMalloc(size); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, size); + return pDst; +} + +static void ctgFreeTableMeta(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneVgroupInfo(void* pSrc) { + SVgroupInfo* pDst = taosMemoryMalloc(sizeof(SVgroupInfo)); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, sizeof(SVgroupInfo)); + return pDst; +} + +static void ctgFreeVgroupInfo(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneTableIndices(void* pSrc) { + return taosArrayDup((const SArray*)pSrc); +} + +static void ctgFreeTableIndices(void* p) { + taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); +} + +static void* ctgCloneFuncInfo(void* pSrc) { + SFuncInfo* pDst = taosMemoryMalloc(sizeof(SFuncInfo)); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, sizeof(SFuncInfo)); + return pDst; +} + +static void ctgFreeFuncInfo(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneIndexInfo(void* pSrc) { + SIndexInfo* pDst = taosMemoryMalloc(sizeof(SIndexInfo)); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, sizeof(SIndexInfo)); + return pDst; +} + +static void ctgFreeIndexInfo(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneUserAuth(void* pSrc) { + bool* pDst = taosMemoryMalloc(sizeof(bool)); + if (NULL == pDst) { + return NULL; + } + *pDst = *(bool*)pSrc; + return pDst; +} + +static void ctgFreeUserAuth(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneQnodeList(void* pSrc) { + return taosArrayDup((const SArray*)pSrc); +} +static void ctgFreeQnodeList(void* p) { + taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); +} + +static void* ctgCloneTableCfg(void* pSrc) { + STableCfg* pDst = taosMemoryMalloc(sizeof(STableCfg)); + if (NULL == pDst) { + return NULL; + } + memcpy(pDst, pSrc, sizeof(STableCfg)); + return pDst; +} + +static void ctgFreeTableCfg(void* p) { + taosMemoryFree(((SMetaRes*)p)->pRes); +} + +static void* ctgCloneDnodeList(void* pSrc) { + return taosArrayDup((const SArray*)pSrc); +} + +static void ctgFreeDnodeList(void* p) { + taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes); +} + +static int32_t ctgCloneMetaDataArray(SArray* pSrc, FCopy copyFunc, SArray** pDst) { + if (NULL == pSrc) { + return TSDB_CODE_SUCCESS; + } + + int32_t size = taosArrayGetSize(pSrc); + *pDst = taosArrayInit(size, sizeof(SMetaRes)); + if (NULL == *pDst) { + return TSDB_CODE_OUT_OF_MEMORY; + } + for (int32_t i = 0; i < size; ++i) { + SMetaRes* pRes = taosArrayGet(pSrc, i); + SMetaRes res = {.code = pRes->code, .pRes = copyFunc(pRes->pRes)}; + if (NULL == res.pRes) { + return TSDB_CODE_OUT_OF_MEMORY; + } + taosArrayPush(*pDst, &res); + } + + return TSDB_CODE_SUCCESS; +} + +SMetaData* catalogCloneMetaData(SMetaData* pData) { + SMetaData* pRes = taosMemoryCalloc(1, sizeof(SMetaData)); + if (NULL == pRes) { + return NULL; + } + + int32_t code = ctgCloneMetaDataArray(pData->pDbVgroup, ctgCloneDbVgroup, &pRes->pDbVgroup); + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pDbCfg, ctgCloneDbCfgInfo, &pRes->pDbCfg); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pDbInfo, ctgCloneDbInfo, &pRes->pDbInfo); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pTableMeta, ctgCloneTableMeta, &pRes->pTableMeta); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pTableHash, ctgCloneVgroupInfo, &pRes->pTableHash); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pTableIndex, ctgCloneTableIndices, &pRes->pTableIndex); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pUdfList, ctgCloneFuncInfo, &pRes->pUdfList); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pIndex, ctgCloneIndexInfo, &pRes->pIndex); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pUser, ctgCloneUserAuth, &pRes->pUser); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pQnodeList, ctgCloneQnodeList, &pRes->pQnodeList); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pTableCfg, ctgCloneTableCfg, &pRes->pTableCfg); + } + if (TSDB_CODE_SUCCESS == code) { + code = ctgCloneMetaDataArray(pData->pDnodeList, ctgCloneDnodeList, &pRes->pDnodeList); + } + + if (TSDB_CODE_SUCCESS != code) { + catalogFreeMetaData(pRes); + return NULL; + } + + return pRes; +} + +void catalogFreeMetaData(SMetaData* pData) { + if (NULL == pData) { + return; + } + + taosArrayDestroyEx(pData->pDbVgroup, ctgFreeDbVgroup); + taosArrayDestroyEx(pData->pDbCfg, ctgFreeDbCfgInfo); + taosArrayDestroyEx(pData->pDbInfo, ctgFreeDbInfo); + taosArrayDestroyEx(pData->pTableMeta, ctgFreeTableMeta); + taosArrayDestroyEx(pData->pTableHash, ctgFreeVgroupInfo); + taosArrayDestroyEx(pData->pTableIndex, ctgFreeTableIndices); + taosArrayDestroyEx(pData->pUdfList, ctgFreeFuncInfo); + taosArrayDestroyEx(pData->pIndex, ctgFreeIndexInfo); + taosArrayDestroyEx(pData->pUser, ctgFreeUserAuth); + taosArrayDestroyEx(pData->pQnodeList, ctgFreeQnodeList); + taosArrayDestroyEx(pData->pTableCfg, ctgFreeTableCfg); + taosArrayDestroyEx(pData->pDnodeList, ctgFreeDnodeList); + taosMemoryFreeClear(pData->pSvrVer); + taosMemoryFree(pData); +} diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c index 4fb55ed373bfdc849853184bd19916c388123ae2..4e698fc999d103038d92bd11bebbf6874d12392f 100644 --- a/source/libs/parser/src/parInsert.c +++ b/source/libs/parser/src/parInsert.c @@ -266,7 +266,7 @@ static int32_t getTableVgroup(SInsertParseContext* pCxt, int32_t tbNo, SName* pT return catalogGetTableHashVgroup(pBasicCtx->pCatalog, &conn, pTbName, pVg); } -static int32_t getTableMetaImpl(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* dbFname, bool isStb) { +static int32_t getTableMetaImpl(SInsertParseContext* pCxt, int32_t tbNo, SName* name, bool isStb) { CHECK_CODE(getTableSchema(pCxt, tbNo, name, isStb, &pCxt->pTableMeta)); if (!isStb) { SVgroupInfo vg; @@ -276,12 +276,12 @@ static int32_t getTableMetaImpl(SInsertParseContext* pCxt, int32_t tbNo, SName* return TSDB_CODE_SUCCESS; } -static int32_t getTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* dbFname) { - return getTableMetaImpl(pCxt, tbNo, name, dbFname, false); +static int32_t getTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name) { + return getTableMetaImpl(pCxt, tbNo, name, false); } -static int32_t getSTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* dbFname) { - return getTableMetaImpl(pCxt, tbNo, name, dbFname, true); +static int32_t getSTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name) { + return getTableMetaImpl(pCxt, tbNo, name, true); } static int32_t getDBCfg(SInsertParseContext* pCxt, const char* pDbFName, SDbCfgInfo* pInfo) { @@ -1178,7 +1178,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, int32_t tbNo, SName* tNameGetFullDbName(&sname, dbFName); strcpy(pCxt->sTableName, sname.tname); - CHECK_CODE(getSTableMeta(pCxt, tbNo, &sname, dbFName)); + CHECK_CODE(getSTableMeta(pCxt, tbNo, &sname)); if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) { return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed"); } @@ -1385,6 +1385,10 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, TdFilePtr fp, STableDataB (*numOfRows)++; } pCxt->pSql = pRawSql; + + if (pDataBlock->nAllocSize > tsMaxMemUsedByInsert * 1024 * 1024) { + break; + } } if (0 == (*numOfRows) && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) { @@ -1393,23 +1397,13 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, TdFilePtr fp, STableDataB return TSDB_CODE_SUCCESS; } -static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STableDataBlocks* dataBuf) { - char filePathStr[TSDB_FILENAME_LEN] = {0}; - if (TK_NK_STRING == filePath.type) { - trimString(filePath.z, filePath.n, filePathStr, sizeof(filePathStr)); - } else { - strncpy(filePathStr, filePath.z, filePath.n); - } - TdFilePtr fp = taosOpenFile(filePathStr, TD_FILE_READ | TD_FILE_STREAM); - if (NULL == fp) { - return TAOS_SYSTEM_ERROR(errno); - } - +static int32_t parseDataFromFileAgain(SInsertParseContext* pCxt, int16_t tableNo, const SName* pTableName, + STableDataBlocks* dataBuf) { int32_t maxNumOfRows; CHECK_CODE(allocateMemIfNeed(dataBuf, getExtendedRowSize(dataBuf), &maxNumOfRows)); int32_t numOfRows = 0; - CHECK_CODE(parseCsvFile(pCxt, fp, dataBuf, maxNumOfRows, &numOfRows)); + CHECK_CODE(parseCsvFile(pCxt, pCxt->pComCxt->csvCxt.fp, dataBuf, maxNumOfRows, &numOfRows)); SSubmitBlk* pBlocks = (SSubmitBlk*)(dataBuf->pData); if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, dataBuf, numOfRows)) { @@ -1417,12 +1411,38 @@ static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STa "too many rows in sql, total number of rows should be less than INT32_MAX"); } + if (!taosEOFFile(pCxt->pComCxt->csvCxt.fp)) { + pCxt->pComCxt->needMultiParse = true; + pCxt->pComCxt->csvCxt.tableNo = tableNo; + memcpy(&pCxt->pComCxt->csvCxt.tableName, pTableName, sizeof(SName)); + pCxt->pComCxt->csvCxt.pLastSqlPos = pCxt->pSql; + } + dataBuf->numOfTables = 1; pCxt->totalNum += numOfRows; return TSDB_CODE_SUCCESS; } +static int32_t parseDataFromFile(SInsertParseContext* pCxt, int16_t tableNo, const SName* pTableName, SToken filePath, + STableDataBlocks* dataBuf) { + char filePathStr[TSDB_FILENAME_LEN] = {0}; + if (TK_NK_STRING == filePath.type) { + trimString(filePath.z, filePath.n, filePathStr, sizeof(filePathStr)); + } else { + strncpy(filePathStr, filePath.z, filePath.n); + } + pCxt->pComCxt->csvCxt.fp = taosOpenFile(filePathStr, TD_FILE_READ | TD_FILE_STREAM); + if (NULL == pCxt->pComCxt->csvCxt.fp) { + return TAOS_SYSTEM_ERROR(errno); + } + + return parseDataFromFileAgain(pCxt, tableNo, pTableName, dataBuf); +} + static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) { + if (!pCxt->pComCxt->needMultiParse) { + taosCloseFile(&pCxt->pComCxt->csvCxt.fp); + } taosMemoryFreeClear(pCxt->pTableMeta); destroyBoundColumnInfo(&pCxt->tags); tdDestroySVCreateTbReq(&pCxt->createTblReq); @@ -1481,7 +1501,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { return buildSyntaxErrMsg(&pCxt->msg, "invalid charactor in SQL", sToken.z); } - if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) { + if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) && + !pCxt->pComCxt->needMultiParse) { return buildInvalidOperationMsg(&pCxt->msg, "no data in sql"); } break; @@ -1536,7 +1557,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { NEXT_TOKEN(pCxt->pSql, sToken); autoCreateTbl = true; } else if (!existedUsing) { - CHECK_CODE(getTableMeta(pCxt, tbNum, &name, dbFName)); + CHECK_CODE(getTableMeta(pCxt, tbNum, &name)); if (TSDB_SUPER_TABLE == pCxt->pTableMeta->tableType) { return buildInvalidOperationMsg(&pCxt->msg, "insert data into super table is not supported"); } @@ -1577,17 +1598,22 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { if (0 == sToken.n || (TK_NK_STRING != sToken.type && TK_NK_ID != sToken.type)) { return buildSyntaxErrMsg(&pCxt->msg, "file path is required following keyword FILE", sToken.z); } - CHECK_CODE(parseDataFromFile(pCxt, sToken, dataBuf)); + CHECK_CODE(parseDataFromFile(pCxt, tbNum, &name, sToken, dataBuf)); pCxt->pOutput->insertType = TSDB_QUERY_TYPE_FILE_INSERT; tbNum++; - continue; + if (!pCxt->pComCxt->needMultiParse) { + continue; + } else { + parserInfo("0x%" PRIx64 " insert from csv. File is too large, do it in batches.", pCxt->pComCxt->requestId); + break; + } } return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is expected", sToken.z); } - qDebug("0x%" PRIx64 " insert input rows: %d", pCxt->pComCxt->requestId, pCxt->totalNum); + parserInfo("0x%" PRIx64 " insert input rows: %d", pCxt->pComCxt->requestId, pCxt->totalNum); if (TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) { SParsedDataColInfo* tags = taosMemoryMalloc(sizeof(pCxt->tags)); @@ -1612,6 +1638,26 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { return buildOutput(pCxt); } +static int32_t parseInsertBodyAgain(SInsertParseContext* pCxt) { + STableDataBlocks* dataBuf = NULL; + CHECK_CODE(getTableMeta(pCxt, pCxt->pComCxt->csvCxt.tableNo, &pCxt->pComCxt->csvCxt.tableName)); + CHECK_CODE(getDataBlockFromList(pCxt->pTableBlockHashObj, &pCxt->pTableMeta->uid, sizeof(pCxt->pTableMeta->uid), + TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk), getTableInfo(pCxt->pTableMeta).rowSize, + pCxt->pTableMeta, &dataBuf, NULL, &pCxt->createTblReq)); + CHECK_CODE(parseDataFromFileAgain(pCxt, pCxt->pComCxt->csvCxt.tableNo, &pCxt->pComCxt->csvCxt.tableName, dataBuf)); + if (taosEOFFile(pCxt->pComCxt->csvCxt.fp)) { + CHECK_CODE(parseInsertBody(pCxt)); + pCxt->pComCxt->needMultiParse = false; + return TSDB_CODE_SUCCESS; + } + parserInfo("0x%" PRIx64 " insert again input rows: %d", pCxt->pComCxt->requestId, pCxt->totalNum); + // merge according to vgId + if (taosHashGetSize(pCxt->pTableBlockHashObj) > 0) { + CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->payloadType, &pCxt->pVgDataBlocks)); + } + return buildOutput(pCxt); +} + // INSERT INTO // tb_name // [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)] @@ -1621,7 +1667,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache) { SInsertParseContext context = { .pComCxt = pContext, - .pSql = (char*)pContext->pSql, + .pSql = pContext->needMultiParse ? (char*)pContext->csvCxt.pLastSqlPos : (char*)pContext->pSql, .msg = {.buf = pContext->pMsg, .len = pContext->msgLen}, .pTableMeta = NULL, .createTblReq = {0}, @@ -1691,10 +1737,16 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache context.pOutput->payloadType = PAYLOAD_TYPE_KV; - int32_t code = skipInsertInto(&context.pSql, &context.msg); - if (TSDB_CODE_SUCCESS == code) { - code = parseInsertBody(&context); + int32_t code = TSDB_CODE_SUCCESS; + if (!context.pComCxt->needMultiParse) { + code = skipInsertInto(&context.pSql, &context.msg); + if (TSDB_CODE_SUCCESS == code) { + code = parseInsertBody(&context); + } + } else { + code = parseInsertBodyAgain(&context); } + if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) { SName* pTable = taosHashIterate(context.pTableNameHashObj, NULL); while (NULL != pTable) { diff --git a/source/libs/parser/src/parser.c b/source/libs/parser/src/parser.c index 2fe6ebfb79447653731ff907e5128dfededdf111..6b3d6c77c1df72200aac5abc6c233a36747f0874 100644 --- a/source/libs/parser/src/parser.c +++ b/source/libs/parser/src/parser.c @@ -214,6 +214,12 @@ int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCata return code; } +void qDestroyParseContext(SParseContext* pCxt) { + taosArrayDestroy(pCxt->pTableMetaPos); + taosArrayDestroy(pCxt->pTableVgroupPos); + taosMemoryFree(pCxt); +} + void qDestroyQuery(SQuery* pQueryNode) { nodesDestroyNode((SNode*)pQueryNode); } int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) { diff --git a/source/libs/parser/test/mockCatalog.cpp b/source/libs/parser/test/mockCatalog.cpp index fcaa5af05ccdb3a9fe1cb8a51177cab679e0dc84..a9360f796c9eee31c45007ebdc365aa2455365fd 100644 --- a/source/libs/parser/test/mockCatalog.cpp +++ b/source/libs/parser/test/mockCatalog.cpp @@ -130,16 +130,16 @@ void generatePerformanceSchema(MockCatalogService* mcs) { * c5 | column | DOUBLE | 8 | */ void generateTestTables(MockCatalogService* mcs, const std::string& db) { - ITableBuilder& builder = mcs->createTableBuilder(db, "t1", TSDB_NORMAL_TABLE, 6) - .setPrecision(TSDB_TIME_PRECISION_MILLI) - .setVgid(1) - .addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP) - .addColumn("c1", TSDB_DATA_TYPE_INT) - .addColumn("c2", TSDB_DATA_TYPE_BINARY, 20) - .addColumn("c3", TSDB_DATA_TYPE_BIGINT) - .addColumn("c4", TSDB_DATA_TYPE_DOUBLE) - .addColumn("c5", TSDB_DATA_TYPE_DOUBLE); - builder.done(); + mcs->createTableBuilder(db, "t1", TSDB_NORMAL_TABLE, 6) + .setPrecision(TSDB_TIME_PRECISION_MILLI) + .setVgid(1) + .addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP) + .addColumn("c1", TSDB_DATA_TYPE_INT) + .addColumn("c2", TSDB_DATA_TYPE_BINARY, 20) + .addColumn("c3", TSDB_DATA_TYPE_BIGINT) + .addColumn("c4", TSDB_DATA_TYPE_DOUBLE) + .addColumn("c5", TSDB_DATA_TYPE_DOUBLE) + .done(); } /* diff --git a/source/libs/parser/test/parTestUtil.cpp b/source/libs/parser/test/parTestUtil.cpp index 14c991917bf534fb2719b2c9d90aff9325042615..bf27fd2e13191cbdae8bdefc478e5f8f0bc86375 100644 --- a/source/libs/parser/test/parTestUtil.cpp +++ b/source/libs/parser/test/parTestUtil.cpp @@ -343,7 +343,6 @@ class ParserTestBaseImpl { unique_ptr query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), destroyQuery); doParseSql(&cxt, query.get()); - SQuery* pQuery = *(query.get()); if (g_dump) { dump();