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

fix(query): set correct last block reader range.

This commit is contained in:
Haojun Liao 2022-09-06 00:26:38 +08:00
parent ef66da9e18
commit e179c10c88
1 changed files with 17 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -235,7 +235,7 @@ static SHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, const STableK
ASSERT(info.lastKey >= pTsdbReader->window.skey && info.lastKey <= pTsdbReader->window.ekey); ASSERT(info.lastKey >= pTsdbReader->window.skey && info.lastKey <= pTsdbReader->window.ekey);
} else { } else {
info.lastKey = pTsdbReader->window.skey; info.lastKey = pTsdbReader->window.ekey;
} }
taosHashPut(pTableMap, &info.uid, sizeof(uint64_t), &info, sizeof(info)); taosHashPut(pTableMap, &info.uid, sizeof(uint64_t), &info, sizeof(info));
@ -1430,6 +1430,10 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
tRowMergerInit(&merge, &fRow, pReader->pSchema); tRowMergerInit(&merge, &fRow, pReader->pSchema);
doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge); doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader, &merge);
TSDBROW fRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
tRowMerge(&merge, &fRow1);
doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge); doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, &merge);
int32_t code = tRowMergerGetRow(&merge, &pTSRow); int32_t code = tRowMergerGetRow(&merge, &pTSRow);
@ -1850,6 +1854,8 @@ static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDum
static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); } static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); }
static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pBlockScanInfo) { static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBlockScanInfo* pBlockScanInfo) {
int32_t step = ASCENDING_TRAVERSE(pLastBlockReader->order)? 1:-1;
while (1) { while (1) {
bool hasVal = tMergeTreeNext(&pLastBlockReader->mergeTree); bool hasVal = tMergeTreeNext(&pLastBlockReader->mergeTree);
if (!hasVal) { if (!hasVal) {
@ -1859,6 +1865,7 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree); TSDBROW row = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
TSDBKEY k = TSDBROW_KEY(&row); TSDBKEY k = TSDBROW_KEY(&row);
if (!hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order)) { if (!hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order)) {
pBlockScanInfo->lastKey = k.ts + step;
return true; return true;
} }
} }
@ -1932,9 +1939,17 @@ static bool initLastBlockReader(SLastBlockReader* pLastBlockReader, STableBlockS
initMemDataIterator(pBlockScanInfo, pReader); initMemDataIterator(pBlockScanInfo, pReader);
pLastBlockReader->uid = pBlockScanInfo->uid; pLastBlockReader->uid = pBlockScanInfo->uid;
STimeWindow w = pLastBlockReader->window;
if (ASCENDING_TRAVERSE(pLastBlockReader->order)) {
w.skey = pBlockScanInfo->lastKey;
} else {
w.ekey = pBlockScanInfo->lastKey;
}
int32_t code = int32_t code =
tMergeTreeOpen(&pLastBlockReader->mergeTree, (pLastBlockReader->order == TSDB_ORDER_DESC), pReader->pFileReader, tMergeTreeOpen(&pLastBlockReader->mergeTree, (pLastBlockReader->order == TSDB_ORDER_DESC), pReader->pFileReader,
pReader->suid, pBlockScanInfo->uid, &pLastBlockReader->window, &pLastBlockReader->verRange); pReader->suid, pBlockScanInfo->uid, &w, &pLastBlockReader->verRange);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return false; return false;
} }