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

[td-11818] select *

This commit is contained in:
Haojun Liao 2022-01-19 16:20:13 +08:00
parent 1832a50932
commit 1145d1a0ea
5 changed files with 23 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -259,7 +259,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryDag* pDag) {
SArray *execNode = taosArrayInit(4, sizeof(SQueryNodeAddr)); SArray *execNode = taosArrayInit(4, sizeof(SQueryNodeAddr));
SQueryNodeAddr addr = {.numOfEps = 1, .inUse = 0, .nodeId = 1}; SQueryNodeAddr addr = {.numOfEps = 1, .inUse = 0, .nodeId = 2};
addr.epAddr[0].port = 6030; addr.epAddr[0].port = 6030;
strcpy(addr.epAddr[0].fqdn, "localhost"); strcpy(addr.epAddr[0].fqdn, "localhost");

20
source/libs/executor/src/executorMain.c
View File

@ -73,12 +73,12 @@ int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_
assert(tsdb != NULL && pSubplan != NULL); assert(tsdb != NULL && pSubplan != NULL);
SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo; SExecTaskInfo** pTask = (SExecTaskInfo**)pTaskInfo;
int32_t code = 0; int32_t code = 0;
uint64_t uid = 0; uint64_t uid = 0;
STimeWindow window = TSWINDOW_INITIALIZER; STimeWindow window = TSWINDOW_INITIALIZER;
int32_t tableType = 0; int32_t tableType = 0;
SPhyNode *pPhyNode = pSubplan->pNode; SPhyNode* pPhyNode = pSubplan->pNode;
STableGroupInfo groupInfo = {0}; STableGroupInfo groupInfo = {0};
int32_t type = pPhyNode->info.type; int32_t type = pPhyNode->info.type;
@ -112,10 +112,10 @@ int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_
} }
} }
code = doCreateExecTaskInfo(pSubplan, pTask, &groupInfo, tsdb); code = doCreateExecTaskInfo(pSubplan, pTask, &groupInfo, tsdb);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
SDataSinkMgtCfg cfg = {.maxDataBlockNum = 1000, .maxDataBlockNumPerQuery = 100}; SDataSinkMgtCfg cfg = {.maxDataBlockNum = 1000, .maxDataBlockNumPerQuery = 100};
code = dsDataSinkMgtInit(&cfg); code = dsDataSinkMgtInit(&cfg);
@ -127,7 +127,7 @@ int32_t qCreateExecTask(void* tsdb, int32_t vgId, SSubplan* pSubplan, qTaskInfo_
*handle = (*pTask)->dsHandle; *handle = (*pTask)->dsHandle;
_error: _error:
// if failed to add ref for all tables in this query, abort current query // if failed to add ref for all tables in this query, abort current query
return code; return code;
} }

7
source/libs/executor/src/executorimpl.c
View File

@ -4952,7 +4952,7 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
} }
static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) { static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) {
SOperatorInfo* pOperator = (SOperatorInfo*) param; SOperatorInfo *pOperator = (SOperatorInfo*) param;
SExchangeInfo *pExchangeInfo = pOperator->info; SExchangeInfo *pExchangeInfo = pOperator->info;
SExecTaskInfo *pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo *pTaskInfo = pOperator->pTaskInfo;
@ -5012,9 +5012,14 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) {
size_t len = pExchangeInfo->pRsp->numOfRows * pColInfoData->info.bytes; size_t len = pExchangeInfo->pRsp->numOfRows * pColInfoData->info.bytes;
memcpy(tmp, pData, len); memcpy(tmp, pData, len);
pColInfoData->pData = tmp;
pData += len; pData += len;
} }
pRes->info.numOfCols = pOperator->numOfOutput;
pRes->info.rows = pExchangeInfo->pRsp->numOfRows;
return pExchangeInfo->pResult; return pExchangeInfo->pResult;
} }

8
source/libs/planner/src/physicalPlanJson.c
View File

@ -559,10 +559,10 @@ static bool timeWindowFromJson(const cJSON* json, void* obj) {
STimeWindow* win = (STimeWindow*)obj; STimeWindow* win = (STimeWindow*)obj;
char* pStartKey = getString(json, jkTimeWindowStartKey); char* pStartKey = getString(json, jkTimeWindowStartKey);
win->skey = strtoll(pStartKey, NULL, 10); win->skey = strtoul(pStartKey, NULL, 10);
char* pEndKey = getString(json, jkTimeWindowEndKey); char* pEndKey = getString(json, jkTimeWindowEndKey);
win->ekey = strtoll(pEndKey, NULL, 10); win->ekey = strtoul(pEndKey, NULL, 10);
tfree(pStartKey); tfree(pStartKey);
tfree(pEndKey); tfree(pEndKey);
@ -783,7 +783,7 @@ static bool nodeAddrFromJson(const cJSON* json, void* obj) {
pSource->taskId = getNumber(json, jkNodeTaskId); pSource->taskId = getNumber(json, jkNodeTaskId);
char* pSchedId = getString(json, jkNodeTaskSchedId); char* pSchedId = getString(json, jkNodeTaskSchedId);
pSource->schedId = strtoll(pSchedId, NULL, 10); pSource->schedId = strtoul(pSchedId, NULL, 10);
tfree(pSchedId); tfree(pSchedId);
bool res = fromObject(json, jkNodeAddr, queryNodeAddrFromJson, &pSource->addr, true); bool res = fromObject(json, jkNodeAddr, queryNodeAddrFromJson, &pSource->addr, true);
@ -1032,7 +1032,7 @@ static bool subplanIdFromJson(const cJSON* json, void* obj) {
SSubplanId* id = (SSubplanId*)obj; SSubplanId* id = (SSubplanId*)obj;
char* queryId = getString(json, jkIdQueryId); char* queryId = getString(json, jkIdQueryId);
id->queryId = strtoll(queryId, NULL, 0); id->queryId = strtoul(queryId, NULL, 0);
tfree(queryId); tfree(queryId);
id->templateId = getNumber(json, jkIdTemplateId); id->templateId = getNumber(json, jkIdTemplateId);

2
source/libs/qworker/src/qworker.c
View File

@ -258,6 +258,8 @@ int32_t qwAddTaskCtxImpl(SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_
char id[sizeof(qId) + sizeof(tId)] = {0}; char id[sizeof(qId) + sizeof(tId)] = {0};
QW_SET_QTID(id, qId, tId); QW_SET_QTID(id, qId, tId);
printf("%"PRIx64", tid:%"PRIx64"\n", qId, tId);
SQWTaskCtx nctx = {0}; SQWTaskCtx nctx = {0};
QW_LOCK(QW_WRITE, &mgmt->ctxLock); QW_LOCK(QW_WRITE, &mgmt->ctxLock);