Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 18fa48ae 编写于 作者: X Xiaoyu Wang
浏览文件
操作

feat: support batch loading of csv files

上级 f298119e
显示空白变更内容
内联 并排
Showing with 563 addition and 206 deletion
include/common/tglobal.h
浏览文件 @ 18fa48ae
......@@ -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[];
......
include/libs/catalog/catalog.h
浏览文件 @ 18fa48ae
......@@ -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
*/
......
include/libs/nodes/querynodes.h
浏览文件 @ 18fa48ae
......@@ -385,7 +385,6 @@ typedef struct SCmdMsgInfo {
SEpSet epSet;
void* pMsg;
int32_t msgLen;
void* pExtension; // todo remove it soon
} SCmdMsgInfo;
typedef enum EQueryExecMode {
......
include/libs/parser/parser.h
浏览文件 @ 18fa48ae
......@@ -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);
......
source/client/inc/clientInt.h
浏览文件 @ 18fa48ae
......@@ -380,9 +380,16 @@ 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);
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);
......@@ -390,6 +397,8 @@ 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
}
......
source/client/src/clientImpl.c
浏览文件 @ 18fa48ae
......@@ -868,9 +868,10 @@ 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;
SSqlCallbackWrapper* pWrapper = param;
SRequestObj* pRequest = pWrapper->pRequest;
STscObj* pTscObj = pRequest->pTscObj;
pRequest->code = code;
......@@ -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,23 +1031,11 @@ 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) {
int32_t code = 0;
pRequest->body.execMode = pQuery->execMode;
switch (pQuery->execMode) {
case QUERY_EXEC_MODE_LOCAL:
asyncExecLocalCmd(pRequest, pQuery);
return;
case QUERY_EXEC_MODE_RPC:
code = asyncExecDdlQuery(pRequest, pQuery);
break;
case QUERY_EXEC_MODE_SCHEDULE: {
SArray* pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));
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),
......@@ -1047,10 +1046,8 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
.pUser = pRequest->pTscObj->user,
.sysInfo = pRequest->pTscObj->sysInfo,
.allocatorId = pRequest->allocatorRefId};
SAppInstInfo* pAppInfo = getAppInfo(pRequest);
SQueryPlan* pDag = NULL;
code = qCreateQueryPlan(&cxt, &pDag, pMnodeList);
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);
......@@ -1059,12 +1056,14 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
}
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};
SRequestConnInfo conn = {.pTrans = getAppInfo(pRequest)->pTransporter,
.requestId = pRequest->requestId,
.requestObjRefId = pRequest->self};
SSchedulerReq req = {
.syncReq = false,
.localReq = (tsQueryPolicy == QUERY_POLICY_CLIENT),
......@@ -1075,7 +1074,7 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
.sql = pRequest->sqlstr,
.startTs = pRequest->metric.start,
.execFp = schedulerExecCb,
.cbParam = pRequest,
.cbParam = pWrapper,
.chkKillFp = chkRequestKilled,
.chkKillParam = (void*)pRequest->self,
.pExecRes = NULL,
......@@ -1085,11 +1084,33 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
} 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);
break;
case QUERY_EXEC_MODE_RPC:
code = asyncExecDdlQuery(pRequest, pQuery);
break;
case QUERY_EXEC_MODE_SCHEDULE: {
code = asyncExecSchQuery(pRequest, pQuery, pResultMeta, pWrapper);
break;
}
case QUERY_EXEC_MODE_EMPTY_RESULT:
......
source/client/src/clientMain.c
浏览文件 @ 18fa48ae
......@@ -667,35 +667,39 @@ 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;
SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)param;
SRequestObj *pRequest = pWrapper->pRequest;
SQuery *pQuery = pRequest->pQuery;
......@@ -703,7 +707,7 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
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;
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);
}
if (TSDB_CODE_SUCCESS == code) {
pWrapper->pParseCtx->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pWrapper->pParseCtx->pCatalog);
}
if (TSDB_CODE_SUCCESS == code) {
pRequest->metric.syntaxStart = taosGetTimestampUs();
SCatalogReq catalogReq = {.forceUpdate = updateMetaForce, .qNodeRequired = qnodeRequired(pRequest)};
code = qParseSqlSyntax(pCxt, &pRequest->pQuery, &catalogReq);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
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;
SRequestConnInfo conn = {.pTrans = pCxt->pTransporter,
.requestId = pCxt->requestId,
.requestObjRefId = pCxt->requestRid,
.mgmtEps = pCxt->mgmtEpSet};
if (TSDB_CODE_SUCCESS == code) {
SRequestConnInfo conn = {.pTrans = pWrapper->pParseCtx->pTransporter,
.requestId = pWrapper->pParseCtx->requestId,
.requestObjRefId = pWrapper->pParseCtx->requestRid,
.mgmtEps = pWrapper->pParseCtx->mgmtEpSet};
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:
if (TSDB_CODE_SUCCESS != code) {
tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code),
pRequest->requestId);
taosMemoryFree(pCxt);
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) {
......
source/client/src/clientSml.c
浏览文件 @ 18fa48ae
......@@ -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;
}
......
source/common/src/tglobal.c
浏览文件 @ 18fa48ae
......@@ -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,7 +502,7 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) {
pItem->stype = stype;
}
/*
/*
pItem = cfgGetItem(tsCfg, "numOfVnodeFetchThreads");
if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) {
tsNumOfVnodeFetchThreads = numOfCores / 4;
......@@ -505,7 +510,7 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) {
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("numOfVnodeWriteThreads", name) == 0) {
tsNumOfVnodeWriteThreads = cfgGetItem(pCfg, "numOfVnodeWriteThreads")->i32;
} else if (strcasecmp("numOfVnodeSyncThreads", name) == 0) {
......
source/libs/catalog/src/ctgUtil.c
浏览文件 @ 18fa48ae
......@@ -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);
}
source/libs/parser/src/parInsert.c
浏览文件 @ 18fa48ae
......@@ -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++;
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);
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) {
......
source/libs/parser/src/parser.c
浏览文件 @ 18fa48ae
......@@ -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) {
......
source/libs/parser/test/mockCatalog.cpp
浏览文件 @ 18fa48ae
......@@ -130,7 +130,7 @@ 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)
mcs->createTableBuilder(db, "t1", TSDB_NORMAL_TABLE, 6)
.setPrecision(TSDB_TIME_PRECISION_MILLI)
.setVgid(1)
.addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
......@@ -138,8 +138,8 @@ void generateTestTables(MockCatalogService* mcs, const std::string& db) {
.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();
.addColumn("c5", TSDB_DATA_TYPE_DOUBLE)
.done();
}
/*
......
source/libs/parser/test/parTestUtil.cpp
浏览文件 @ 18fa48ae
......@@ -343,7 +343,6 @@ class ParserTestBaseImpl {
unique_ptr<SQuery*, void (*)(SQuery**)> query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), destroyQuery);
doParseSql(&cxt, query.get());
SQuery* pQuery = *(query.get());
if (g_dump) {
dump();
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或