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

fix(query): set correct status.

This commit is contained in:
Haojun Liao 2022-11-08 00:23:40 +08:00
parent b16853983e
commit 1da67a569d
2 changed files with 34 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1853,21 +1853,32 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
int64_t startTs = taosGetTimestampUs(); int64_t startTs = taosGetTimestampUs();
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources); size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
int32_t completed = 0;
for (int32_t k = 0; k < totalSources; ++k) {
SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k);
if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
completed += 1;
}
}
if (completed == totalSources) {
setAllSourcesCompleted(pOperator, startTs);
return;
}
while (1) { while (1) {
// printf("1\n"); // printf("1\n");
tsem_wait(&pExchangeInfo->ready); tsem_wait(&pExchangeInfo->ready);
// printf("2\n"); // printf("2\n");
// int32_t completed = 0;
for (int32_t i = 0; i < totalSources; ++i) { for (int32_t i = 0; i < totalSources; ++i) {
SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i); SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i);
if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) { if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
// printf("========:%d is completed\n", i); // printf("========:%d is completed\n", i);
// completed += 1;
continue; continue;
} }
// printf("index:%d --------3\n", i); // printf("index:%d - status:%d\n", i, pDataInfo->status);
if (pDataInfo->status != EX_SOURCE_DATA_READY) { if (pDataInfo->status != EX_SOURCE_DATA_READY) {
// printf("-----------%d, status:%d, continue\n", i, pDataInfo->status); // printf("-----------%d, status:%d, continue\n", i, pDataInfo->status);
continue; continue;
@ -1885,33 +1896,27 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo; SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
if (pRsp->numOfRows == 0) { if (pRsp->numOfRows == 0) {
pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
// printf("%d completed, try next\n", i);
int32_t completed = 0;
for (int32_t k = 0; k < totalSources; ++k) {
SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k);
if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
completed += 1;
}
}
qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64 qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 ", completed:%d try next %d/%" PRIzu, ", totalRows:%" PRIu64 ", completed:%d try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows, GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows,
pExchangeInfo->loadInfo.totalRows, completed + 1, i + 1, totalSources); pExchangeInfo->loadInfo.totalRows, completed, i + 1, totalSources);
taosMemoryFreeClear(pDataInfo->pRsp); taosMemoryFreeClear(pDataInfo->pRsp);
if (completed == totalSources) { // if (completed == totalSources) {
setAllSourcesCompleted(pOperator, startTs); // return;
return; // } else {
} else { // break;
// }
break; break;
} }
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp; SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
int32_t index = 0; int32_t index = 0;
char* pStart = pRetrieveRsp->data; char* pStart = pRetrieveRsp->data;
while (index++ < pRetrieveRsp->numOfBlocks) { while (index++ < pRetrieveRsp->numOfBlocks) {
printf("results, numOfBLock: %d\n", pRetrieveRsp->numOfBlocks);
SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false); SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart); code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
if (code != 0) { if (code != 0) {
@ -1924,16 +1929,16 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator); updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator);
int32_t completed = 0; // int32_t completed = 0;
if (pRsp->completed == 1) { if (pRsp->completed == 1) {
pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED; pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
for (int32_t k = 0; k < totalSources; ++k) { // for (int32_t k = 0; k < totalSources; ++k) {
SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k); // SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k);
if (p->status == EX_SOURCE_DATA_EXHAUSTED) { // if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
completed += 1; // completed += 1;
} // }
} // }
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
" execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64 " execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
@ -1959,9 +1964,9 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
} }
} }
if (completed == totalSources) { // if (completed == totalSources) {
setAllSourcesCompleted(pOperator, startTs); // setAllSourcesCompleted(pOperator, startTs);
} // }
return; return;
} }
@ -1975,7 +1980,6 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
} }
if (completed == totalSources) { if (completed == totalSources) {
setAllSourcesCompleted(pOperator, startTs);
return; return;
} }
} }

1
tests/script/tsim/parser/auto_create_tb.sim
View File

@ -186,6 +186,7 @@ endi
sql select t1, count(*), first(c9) from $stb partition by t1 order by t1 asc slimit 3 sql select t1, count(*), first(c9) from $stb partition by t1 order by t1 asc slimit 3
if $rows != 3 then if $rows != 3 then
print expect 3, actual: $rows
return -1 return -1
endi endi
if $data(1)[1] != 1 then if $data(1)[1] != 1 then