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

fix(query): set update ts flag for stream.

This commit is contained in:
Haojun Liao 2022-12-13 00:16:40 +08:00
parent d7edcfd207
commit e3e42cdfbd
5 changed files with 13 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
source/common/src/tdatablock.c
View File

@ -358,6 +358,10 @@ size_t blockDataGetNumOfCols(const SSDataBlock* pBlock) { return taosArrayGetSiz
size_t blockDataGetNumOfRows(const SSDataBlock* pBlock) { return pBlock->info.rows; } size_t blockDataGetNumOfRows(const SSDataBlock* pBlock) { return pBlock->info.rows; }
int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex) { int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex) {
if (pDataBlock->info.rows > 0) {
ASSERT(pDataBlock->info.dataLoad == 1);
}
if (pDataBlock == NULL || pDataBlock->info.rows <= 0 || pDataBlock->info.dataLoad == 0) { if (pDataBlock == NULL || pDataBlock->info.rows <= 0 || pDataBlock->info.dataLoad == 0) {
return 0; return 0;
} }

1
source/dnode/vnode/src/tq/tqRead.c
View File

@ -533,6 +533,7 @@ int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReader* pReader) {
pBlock->info.id.uid = pReader->msgIter.uid; pBlock->info.id.uid = pReader->msgIter.uid;
pBlock->info.rows = pReader->msgIter.numOfRows; pBlock->info.rows = pReader->msgIter.numOfRows;
pBlock->info.version = pReader->pMsg->version; pBlock->info.version = pReader->pMsg->version;
pBlock->info.dataLoad = 1;
while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) { while ((row = tGetSubmitBlkNext(&pReader->blkIter)) != NULL) {
tdSTSRowIterReset(&iter, row); tdSTSRowIterReset(&iter, row);

1
source/libs/executor/src/exchangeoperator.c
View File

@ -510,6 +510,7 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pCo
blockDataEnsureCapacity(pRes, pBlock->info.rows); blockDataEnsureCapacity(pRes, pBlock->info.rows);
// data from mnode // data from mnode
pRes->info.dataLoad = 1;
pRes->info.rows = pBlock->info.rows; pRes->info.rows = pBlock->info.rows;
relocateColumnData(pRes, pColList, pBlock->pDataBlock, false); relocateColumnData(pRes, pColList, pBlock->pDataBlock, false);
blockDataDestroy(pBlock); blockDataDestroy(pBlock);

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

@ -2546,6 +2546,7 @@ int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pStat
pBlock->info.rows += pRow->numOfRows; pBlock->info.rows += pRow->numOfRows;
releaseOutputBuf(pState, &key, pRow); releaseOutputBuf(pState, &key, pRow);
} }
pBlock->info.dataLoad = 1;
blockDataUpdateTsWindow(pBlock, 0); blockDataUpdateTsWindow(pBlock, 0);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -2635,6 +2636,7 @@ int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pSta
} }
} }
pBlock->info.dataLoad = 1;
pBlock->info.rows += pRow->numOfRows; pBlock->info.rows += pRow->numOfRows;
// saveSessionDiscBuf(pState, pKey, pVal, size); // saveSessionDiscBuf(pState, pKey, pVal, size);
releaseOutputBuf(pState, NULL, pRow); releaseOutputBuf(pState, NULL, pRow);

5
source/libs/executor/src/scanoperator.c
View File

@ -1313,6 +1313,7 @@ static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock,
} }
pDestBlock->info.type = STREAM_CLEAR; pDestBlock->info.type = STREAM_CLEAR;
pDestBlock->info.version = pSrcBlock->info.version; pDestBlock->info.version = pSrcBlock->info.version;
pDestBlock->info.dataLoad = 1;
blockDataUpdateTsWindow(pDestBlock, 0); blockDataUpdateTsWindow(pDestBlock, 0);
return code; return code;
} }
@ -1421,6 +1422,7 @@ static void checkUpdateData(SStreamScanInfo* pInfo, bool invertible, SSDataBlock
} }
if (out && pInfo->pUpdateDataRes->info.rows > 0) { if (out && pInfo->pUpdateDataRes->info.rows > 0) {
pInfo->pUpdateDataRes->info.version = pBlock->info.version; pInfo->pUpdateDataRes->info.version = pBlock->info.version;
pInfo->pUpdateDataRes->info.dataLoad = 1;
blockDataUpdateTsWindow(pInfo->pUpdateDataRes, 0); blockDataUpdateTsWindow(pInfo->pUpdateDataRes, 0);
pInfo->pUpdateDataRes->info.type = pInfo->partitionSup.needCalc ? STREAM_DELETE_DATA : STREAM_CLEAR; pInfo->pUpdateDataRes->info.type = pInfo->partitionSup.needCalc ? STREAM_DELETE_DATA : STREAM_CLEAR;
} }
@ -1483,6 +1485,7 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL); doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
} }
pInfo->pRes->info.dataLoad = 1;
blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex); blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
blockDataFreeRes((SSDataBlock*)pBlock); blockDataFreeRes((SSDataBlock*)pBlock);
@ -1771,6 +1774,7 @@ FETCH_NEXT_BLOCK:
// TODO move into scan // TODO move into scan
pBlock->info.calWin.skey = INT64_MIN; pBlock->info.calWin.skey = INT64_MIN;
pBlock->info.calWin.ekey = INT64_MAX; pBlock->info.calWin.ekey = INT64_MAX;
pBlock->info.dataLoad = 1;
blockDataUpdateTsWindow(pBlock, 0); blockDataUpdateTsWindow(pBlock, 0);
switch (pBlock->info.type) { switch (pBlock->info.type) {
case STREAM_NORMAL: case STREAM_NORMAL:
@ -1948,6 +1952,7 @@ FETCH_NEXT_BLOCK:
} }
doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL); doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
pInfo->pRes->info.dataLoad = 1;
blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex); blockDataUpdateTsWindow(pInfo->pRes, pInfo->primaryTsIndex);
if (pBlockInfo->rows > 0 || pInfo->pUpdateDataRes->info.rows > 0) { if (pBlockInfo->rows > 0 || pInfo->pUpdateDataRes->info.rows > 0) {