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

Merge pull request #7463 from taosdata/feature/query 

Feature/query
This commit is contained in:
Haojun Liao 2021-08-19 16:22:17 +08:00 committed by GitHub
parent 619d9712c6 ac3208736b
commit 4c2e4f2930
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 52 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
src/client/src/tscSubquery.c
View File

@ -2433,6 +2433,26 @@ int32_t tscHandleFirstRoundStableQuery(SSqlObj *pSql) {
return terrno; return terrno;
} }
typedef struct SPair {
int32_t first;
int32_t second;
} SPair;
static void doSendQueryReqs(SSchedMsg* pSchedMsg) {
SSqlObj* pSql = pSchedMsg->ahandle;
SPair* p = pSchedMsg->msg;
for(int32_t i = p->first; i < p->second; ++i) {
SSqlObj* pSub = pSql->pSubs[i];
SRetrieveSupport* pSupport = pSub->param;
tscDebug("0x%"PRIx64" sub:0x%"PRIx64" launch subquery, orderOfSub:%d.", pSql->self, pSub->self, pSupport->subqueryIndex);
tscBuildAndSendRequest(pSub, NULL);
}
tfree(p);
}
int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) { int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
SSqlRes *pRes = &pSql->res; SSqlRes *pRes = &pSql->res;
SSqlCmd *pCmd = &pSql->cmd; SSqlCmd *pCmd = &pSql->cmd;
@ -2556,12 +2576,32 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
return pRes->code; return pRes->code;
} }
for(int32_t j = 0; j < pState->numOfSub; ++j) { // concurrently sent the query requests.
SSqlObj* pSub = pSql->pSubs[j]; const int32_t MAX_REQUEST_PER_TASK = 8;
SRetrieveSupport* pSupport = pSub->param;
tscDebug("0x%"PRIx64" sub:0x%"PRIx64" launch subquery, orderOfSub:%d.", pSql->self, pSub->self, pSupport->subqueryIndex); int32_t numOfTasks = (pState->numOfSub + MAX_REQUEST_PER_TASK - 1)/MAX_REQUEST_PER_TASK;
tscBuildAndSendRequest(pSub, NULL); assert(numOfTasks >= 1);
int32_t num = (pState->numOfSub/numOfTasks) + 1;
tscDebug("0x%"PRIx64 " query will be sent by %d threads", pSql->self, numOfTasks);
for(int32_t j = 0; j < numOfTasks; ++j) {
SSchedMsg schedMsg = {0};
schedMsg.fp = doSendQueryReqs;
schedMsg.ahandle = (void*)pSql;
schedMsg.thandle = NULL;
SPair* p = calloc(1, sizeof(SPair));
p->first = j * num;
if (j == numOfTasks - 1) {
p->second = pState->numOfSub;
} else {
p->second = (j + 1) * num;
}
schedMsg.msg = p;
taosScheduleTask(tscQhandle, &schedMsg);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;

5
src/query/src/qExecutor.c
View File

@ -4265,6 +4265,7 @@ static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data
} }
qDebug("QInfo:0x%"PRIx64" set %d subscribe info", pQInfo->qId, total); qDebug("QInfo:0x%"PRIx64" set %d subscribe info", pQInfo->qId, total);
// Check if query is completed or not for stable query or normal table query respectively. // Check if query is completed or not for stable query or normal table query respectively.
if (Q_STATUS_EQUAL(pRuntimeEnv->status, QUERY_COMPLETED) && pRuntimeEnv->proot->status == OP_EXEC_DONE) { if (Q_STATUS_EQUAL(pRuntimeEnv->status, QUERY_COMPLETED) && pRuntimeEnv->proot->status == OP_EXEC_DONE) {
setQueryStatus(pRuntimeEnv, QUERY_OVER); setQueryStatus(pRuntimeEnv, QUERY_OVER);
@ -7079,6 +7080,10 @@ static SSDataBlock* doTagScan(void* param, bool* newgroup) {
qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_QID(pRuntimeEnv), count); qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_QID(pRuntimeEnv), count);
} }
if (pOperator->status == OP_EXEC_DONE) {
setQueryStatus(pOperator->pRuntimeEnv, QUERY_COMPLETED);
}
pRes->info.rows = count; pRes->info.rows = count;
return (pRes->info.rows == 0)? NULL:pInfo->pRes; return (pRes->info.rows == 0)? NULL:pInfo->pRes;
} }