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

enh: insert parser refactor

This commit is contained in:
Xiaoyu Wang 2022-10-24 16:58:45 +08:00
parent 2aed458bf7
commit 292f0c6db7
3 changed files with 15 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/client/inc/clientInt.h
View File

@ -46,7 +46,6 @@ extern "C" {
#define ERROR_MSG_BUF_DEFAULT_SIZE 512 #define ERROR_MSG_BUF_DEFAULT_SIZE 512
#define HEARTBEAT_INTERVAL 1500 // ms #define HEARTBEAT_INTERVAL 1500 // ms
#define SYNC_ON_TOP_OF_ASYNC 1
enum { enum {
RES_TYPE__QUERY = 1, RES_TYPE__QUERY = 1,
@ -271,7 +270,6 @@ void doFreeReqResultInfo(SReqResultInfo* pResInfo);
int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName, SArray** pReq); int32_t transferTableNameList(const char* tbList, int32_t acctId, char* dbName, SArray** pReq);
void syncCatalogFn(SMetaData* pResult, void* param, int32_t code); void syncCatalogFn(SMetaData* pResult, void* param, int32_t code);
SRequestObj* execQuery(uint64_t connId, const char* sql, int sqlLen, bool validateOnly);
TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly); TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly);
void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly); void taosAsyncQueryImpl(uint64_t connId, const char* sql, __taos_async_fn_t fp, void* param, bool validateOnly);
@ -349,8 +347,6 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet);
STscObj* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db, STscObj* taos_connect_internal(const char* ip, const char* user, const char* pass, const char* auth, const char* db,
uint16_t port, int connType); uint16_t port, int connType);
SRequestObj* launchQuery(uint64_t connId, const char* sql, int sqlLen, bool validateOnly, bool inRetry);
int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb); int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtCallback* pStmtCb);
int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList); int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList);

60
source/client/src/clientImpl.c
View File

@ -1013,28 +1013,6 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQue
return pRequest; return pRequest;
} }
SRequestObj* launchQuery(uint64_t connId, const char* sql, int sqlLen, bool validateOnly, bool inRetry) {
SRequestObj* pRequest = NULL;
SQuery* pQuery = NULL;
int32_t code = buildRequest(connId, sql, sqlLen, NULL, validateOnly, &pRequest);
if (code != TSDB_CODE_SUCCESS) {
terrno = code;
return NULL;
}
code = parseSql(pRequest, false, &pQuery, NULL);
if (code != TSDB_CODE_SUCCESS) {
pRequest->code = code;
return pRequest;
}
pRequest->inRetry = inRetry;
pRequest->stableQuery = pQuery->stableQuery;
return launchQueryImpl(pRequest, pQuery, false, NULL);
}
static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta,
SSqlCallbackWrapper* pWrapper) { SSqlCallbackWrapper* pWrapper) {
pRequest->type = pQuery->msgType; pRequest->type = pQuery->msgType;
@ -1197,32 +1175,6 @@ int32_t removeMeta(STscObj* pTscObj, SArray* tbList) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
// todo remove it soon
SRequestObj* execQuery(uint64_t connId, const char* sql, int sqlLen, bool validateOnly) {
SRequestObj* pRequest = NULL;
int32_t retryNum = 0;
int32_t code = 0;
bool inRetry = false;
do {
destroyRequest(pRequest);
pRequest = launchQuery(connId, sql, sqlLen, validateOnly, inRetry);
if (pRequest == NULL || TSDB_CODE_SUCCESS == pRequest->code || !NEED_CLIENT_HANDLE_ERROR(pRequest->code)) {
break;
}
code = refreshMeta(pRequest->pTscObj, pRequest);
if (code) {
pRequest->code = code;
break;
}
inRetry = true;
} while (retryNum++ < REQUEST_TOTAL_EXEC_TIMES);
return pRequest;
}
int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) { int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) {
pEpSet->version = 0; pEpSet->version = 0;
@ -2291,7 +2243,6 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) {
return NULL; return NULL;
} }
#if SYNC_ON_TOP_OF_ASYNC
SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam)); SSyncQueryParam* param = taosMemoryCalloc(1, sizeof(SSyncQueryParam));
tsem_init(&param->sem, 0, 0); tsem_init(&param->sem, 0, 0);
@ -2301,15 +2252,4 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) {
param->pRequest->syncQuery = true; param->pRequest->syncQuery = true;
} }
return param->pRequest; return param->pRequest;
#else
size_t sqlLen = strlen(sql);
if (sqlLen > (size_t)TSDB_MAX_ALLOWED_SQL_LEN) {
tscError("sql string exceeds max length:%d", TSDB_MAX_ALLOWED_SQL_LEN);
terrno = TSDB_CODE_TSC_EXCEED_SQL_LIMIT;
return NULL;
}
TAOS_RES* pRes = execQuery(*(int64_t*)taos, sql, sqlLen, validateOnly);
return pRes;
#endif
} }

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

@ -261,12 +261,7 @@ TAOS_ROW taos_fetch_row(TAOS_RES *res) {
return NULL; return NULL;
} }
#if SYNC_ON_TOP_OF_ASYNC
return doAsyncFetchRows(pRequest, true, true); return doAsyncFetchRows(pRequest, true, true);
#else
return doFetchRows(pRequest, true, true);
#endif
} else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) { } else if (TD_RES_TMQ(res) || TD_RES_TMQ_METADATA(res)) {
SMqRspObj *msg = ((SMqRspObj *)res); SMqRspObj *msg = ((SMqRspObj *)res);
SReqResultInfo *pResultInfo; SReqResultInfo *pResultInfo;
@ -549,11 +544,7 @@ int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows) {
return 0; return 0;
} }
#if SYNC_ON_TOP_OF_ASYNC
doAsyncFetchRows(pRequest, false, true); doAsyncFetchRows(pRequest, false, true);
#else
doFetchRows(pRequest, true, true);
#endif
// TODO refactor // TODO refactor
SReqResultInfo *pResultInfo = &pRequest->body.resInfo; SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
@ -601,11 +592,7 @@ int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData) {
return 0; return 0;
} }
#if SYNC_ON_TOP_OF_ASYNC
doAsyncFetchRows(pRequest, false, false); doAsyncFetchRows(pRequest, false, false);
#else
doFetchRows(pRequest, false, false);
#endif
SReqResultInfo *pResultInfo = &pRequest->body.resInfo; SReqResultInfo *pResultInfo = &pRequest->body.resInfo;