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

fix(query): fix bug in query last block.

This commit is contained in:
Haojun Liao 2022-08-18 18:48:50 +08:00
parent 2cd74016f7
commit e75c400045
1 changed files with 93 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -89,7 +89,7 @@ typedef struct SLastBlockReader {
STimeWindow window; STimeWindow window;
SVersionRange verRange; SVersionRange verRange;
uint64_t uid; uint64_t uid;
int32_t rowIndex; int16_t* rowIndex; // row index ptr, usually from the STableBlockScanInfo->indexInBlockL
} SLastBlockReader; } SLastBlockReader;
typedef struct SFilesetIter { typedef struct SFilesetIter {
@ -181,7 +181,8 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* re
int8_t* pLevel); int8_t* pLevel);
static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level); static SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level);
static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader); static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader);
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader); static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader);
static int32_t doBuildDataBlock(STsdbReader* pReader);
static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) { static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo; SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
@ -449,7 +450,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd
pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level); pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level);
pReader->suid = pCond->suid; pReader->suid = pCond->suid;
pReader->order = pCond->order; pReader->order = pCond->order;
pReader->capacity = capacity; pReader->capacity = 1;
pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL; pReader->idStr = (idstr != NULL) ? strdup(idstr) : NULL;
pReader->verRange = getQueryVerRange(pVnode, pCond, level); pReader->verRange = getQueryVerRange(pVnode, pCond, level);
pReader->type = pCond->type; pReader->type = pCond->type;
@ -624,24 +625,29 @@ _end:
return code; return code;
} }
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SArray* pLastBlockIndex, static void cleanupTableScanInfo(SHashObj* pTableMap) {
SBlockNumber * pBlockNum, SArray* pQualifiedLastBlock) {
int32_t numOfQTable= 0;
size_t numOfTables = taosArrayGetSize(pIndexList);
int64_t st = taosGetTimestampUs();
size_t size = 0;
STableBlockScanInfo* px = NULL; STableBlockScanInfo* px = NULL;
while (1) { while (1) {
px = taosHashIterate(pReader->status.pTableMap, px); px = taosHashIterate(pTableMap, px);
if (px == NULL) { if (px == NULL) {
break; break;
} }
// reset the index in last block when handing a new file
px->indexInBlockL = -1;
tMapDataClear(&px->mapData); tMapDataClear(&px->mapData);
taosArrayClear(px->pBlockList); taosArrayClear(px->pBlockList);
} }
}
static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SArray* pLastBlockIndex,
SBlockNumber * pBlockNum, SArray* pQualifiedLastBlock) {
int32_t numOfQTable = 0;
size_t sizeInDisk = 0;
size_t numOfTables = taosArrayGetSize(pIndexList);
int64_t st = taosGetTimestampUs();
cleanupTableScanInfo(pReader->status.pTableMap);
for (int32_t i = 0; i < numOfTables; ++i) { for (int32_t i = 0; i < numOfTables; ++i) {
SBlockIdx* pBlockIdx = taosArrayGet(pIndexList, i); SBlockIdx* pBlockIdx = taosArrayGet(pIndexList, i);
@ -651,7 +657,7 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SArray*
tMapDataReset(&pScanInfo->mapData); tMapDataReset(&pScanInfo->mapData);
tsdbReadBlock(pReader->pFileReader, pBlockIdx, &pScanInfo->mapData); tsdbReadBlock(pReader->pFileReader, pBlockIdx, &pScanInfo->mapData);
size += pScanInfo->mapData.nData; sizeInDisk += pScanInfo->mapData.nData;
for (int32_t j = 0; j < pScanInfo->mapData.nItem; ++j) { for (int32_t j = 0; j < pScanInfo->mapData.nItem; ++j) {
SBlock block = {0}; SBlock block = {0};
tMapDataGetItemByIdx(&pScanInfo->mapData, j, &block, tGetBlock); tMapDataGetItemByIdx(&pScanInfo->mapData, j, &block, tGetBlock);
@ -707,7 +713,8 @@ static int32_t doLoadFileBlock(STsdbReader* pReader, SArray* pIndexList, SArray*
double el = (taosGetTimestampUs() - st) / 1000.0; double el = (taosGetTimestampUs() - st) / 1000.0;
tsdbDebug("load block of %d tables completed, blocks:%d in %d tables, lastBlock:%d, size:%.2f Kb, elapsed time:%.2f ms %s", tsdbDebug("load block of %d tables completed, blocks:%d in %d tables, lastBlock:%d, size:%.2f Kb, elapsed time:%.2f ms %s",
numOfTables, total, numOfQTable, pBlockNum->numOfLastBlocks, size / 1000.0, el, pReader->idStr); numOfTables, total, numOfQTable, pBlockNum->numOfLastBlocks, sizeInDisk
/ 1000.0, el, pReader->idStr);
pReader->cost.numOfBlocks += total; pReader->cost.numOfBlocks += total;
pReader->cost.headFileLoadTime += el; pReader->cost.headFileLoadTime += el;
@ -1396,7 +1403,7 @@ static int32_t doMergeBufAndFileRows_Rv(STsdbReader* pReader, STableBlockScanInf
if (minKey == tsLast) { if (minKey == tsLast) {
if (!init) { if (!init) {
init = true; init = true;
TSDBROW fRow1 = tsdbRowFromBlockData(pLastBlockData, pLastBlockReader->rowIndex); TSDBROW fRow1 = tsdbRowFromBlockData(pLastBlockData, *pLastBlockReader->rowIndex);
tRowMergerInit(&merge, &fRow1, pReader->pSchema); tRowMergerInit(&merge, &fRow1, pReader->pSchema);
} }
@ -1541,7 +1548,7 @@ static int32_t doMergeThreeLevelRowsRv(STsdbReader* pReader, STableBlockScanInfo
if (minKey == tsLast) { if (minKey == tsLast) {
if (!init) { if (!init) {
init = true; init = true;
TSDBROW fRow1 = tsdbRowFromBlockData(pLastBlockData, pLastBlockReader->rowIndex); TSDBROW fRow1 = tsdbRowFromBlockData(pLastBlockData, *pLastBlockReader->rowIndex);
tRowMergerInit(&merge, &fRow1, pReader->pSchema); tRowMergerInit(&merge, &fRow1, pReader->pSchema);
} }
doMergeRowsInLastBlock(pLastBlockReader, tsLast, &merge); doMergeRowsInLastBlock(pLastBlockReader, tsLast, &merge);
@ -1747,7 +1754,7 @@ 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 void initLastBlockReader(SLastBlockReader* pLastBlockReader, uint64_t uid, STimeWindow* pWin, static void initLastBlockReader(SLastBlockReader* pLastBlockReader, uint64_t uid, STimeWindow* pWin,
SVersionRange* pVerRange, int16_t startPos) { SVersionRange* pVerRange, int16_t* startPos) {
pLastBlockReader->uid = uid; pLastBlockReader->uid = uid;
pLastBlockReader->window = *pWin; pLastBlockReader->window = *pWin;
pLastBlockReader->verRange = *pVerRange; pLastBlockReader->verRange = *pVerRange;
@ -1755,14 +1762,14 @@ static void initLastBlockReader(SLastBlockReader* pLastBlockReader, uint64_t uid
} }
static bool nextRowInLastBlock(SLastBlockReader *pLastBlockReader) { static bool nextRowInLastBlock(SLastBlockReader *pLastBlockReader) {
if (pLastBlockReader->rowIndex >= pLastBlockReader->lastBlockData.nRow) { if (*(pLastBlockReader->rowIndex) >= pLastBlockReader->lastBlockData.nRow) {
return false; return false;
} }
pLastBlockReader->rowIndex += 1; *(pLastBlockReader->rowIndex) += 1;
SBlockData* pBlockData = &pLastBlockReader->lastBlockData; SBlockData* pBlockData = &pLastBlockReader->lastBlockData;
for(int32_t i = pLastBlockReader->rowIndex; i < pBlockData->nRow; ++i) { for(int32_t i = *(pLastBlockReader->rowIndex); i < pBlockData->nRow; ++i) {
if (pBlockData->aUid[i] != pLastBlockReader->uid) { if (pBlockData->aUid[i] != pLastBlockReader->uid) {
continue; continue;
} }
@ -1777,23 +1784,25 @@ static bool nextRowInLastBlock(SLastBlockReader *pLastBlockReader) {
// no data any more // no data any more
if (pBlockData->aTSKEY[i] > pLastBlockReader->window.ekey) { if (pBlockData->aTSKEY[i] > pLastBlockReader->window.ekey) {
pLastBlockReader->rowIndex = pBlockData->nRow; *(pLastBlockReader->rowIndex) = pBlockData->nRow;
return false; return false;
} }
if (pBlockData->aVersion[i] > pLastBlockReader->verRange.maxVer) { if (pBlockData->aVersion[i] > pLastBlockReader->verRange.maxVer) {
pLastBlockReader->rowIndex = pBlockData->nRow; *(pLastBlockReader->rowIndex) = pBlockData->nRow;
return false; return false;
} }
pLastBlockReader->rowIndex = i; *(pLastBlockReader->rowIndex) = i;
return true; return true;
} }
pLastBlockReader->rowIndex = pBlockData->nRow; // set all data is consumed in last block
*(pLastBlockReader->rowIndex) = pBlockData->nRow;
return false; return false;
} }
#if 0
static int32_t saveCurrentState(SLastBlockReader* pLastBlockReader) { static int32_t saveCurrentState(SLastBlockReader* pLastBlockReader) {
return pLastBlockReader->rowIndex; return pLastBlockReader->rowIndex;
} }
@ -1801,15 +1810,16 @@ static int32_t saveCurrentState(SLastBlockReader* pLastBlockReader) {
static void restoreState(SLastBlockReader* pLastBlockReader, int32_t state) { static void restoreState(SLastBlockReader* pLastBlockReader, int32_t state) {
pLastBlockReader->rowIndex = state; pLastBlockReader->rowIndex = state;
} }
#endif
static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) { static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
SBlockData* pBlockData = &pLastBlockReader->lastBlockData; SBlockData* pBlockData = &pLastBlockReader->lastBlockData;
return pBlockData->aTSKEY[pLastBlockReader->rowIndex]; return pBlockData->aTSKEY[*pLastBlockReader->rowIndex];
} }
// todo handle desc order // todo handle desc order
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) {
if (pLastBlockReader->rowIndex >= pLastBlockReader->lastBlockData.nRow) { if (*pLastBlockReader->rowIndex >= pLastBlockReader->lastBlockData.nRow) {
return false; return false;
} }
@ -1847,7 +1857,7 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanI
STSRow* pTSRow = NULL; STSRow* pTSRow = NULL;
SRowMerger merge = {0}; SRowMerger merge = {0};
TSDBROW fRow1 = tsdbRowFromBlockData(pLastBlockData, pLastBlockReader->rowIndex); TSDBROW fRow1 = tsdbRowFromBlockData(pLastBlockData, *pLastBlockReader->rowIndex);
tRowMergerInit(&merge, &fRow1, pReader->pSchema); tRowMergerInit(&merge, &fRow1, pReader->pSchema);
doMergeRowsInLastBlock(pLastBlockReader, ts, &merge); doMergeRowsInLastBlock(pLastBlockReader, ts, &merge);
@ -1898,7 +1908,7 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanI
STSRow* pTSRow = NULL; STSRow* pTSRow = NULL;
SRowMerger merge = {0}; SRowMerger merge = {0};
TSDBROW fRow = tsdbRowFromBlockData(pLastBlockData, pLastBlockReader->rowIndex); TSDBROW fRow = tsdbRowFromBlockData(pLastBlockData, *pLastBlockReader->rowIndex);
tRowMergerInit(&merge, &fRow, pReader->pSchema); tRowMergerInit(&merge, &fRow, pReader->pSchema);
doMergeRowsInLastBlock(pLastBlockReader, tsLastBlock, &merge); doMergeRowsInLastBlock(pLastBlockReader, tsLastBlock, &merge);
@ -1925,12 +1935,9 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
pBlockScanInfo = pReader->status.pTableIter; pBlockScanInfo = pReader->status.pTableIter;
} }
SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader; SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader;
initLastBlockReader(pLastBlockReader, pBlockScanInfo->uid, &pReader->window, &pReader->verRange, &pBlockScanInfo->indexInBlockL);
int16_t startIndex = pBlockInfo != NULL? pBlockScanInfo->indexInBlockL:-1; // bool has = nextRowInLastBlock(pLastBlockReader); // todo handle if no data in last block
initLastBlockReader(pLastBlockReader, pBlockScanInfo->uid, &pReader->window, &pReader->verRange, startIndex);
bool has = nextRowInLastBlock(pLastBlockReader); // todo handle if no data in last block
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
SBlockData* pBlockData = &pReader->status.fileBlockData; SBlockData* pBlockData = &pReader->status.fileBlockData;
@ -1961,7 +1968,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
buildComposedDataBlockImpl(pReader, pBlockScanInfo, pBlockData, pLastBlockReader); buildComposedDataBlockImpl(pReader, pBlockScanInfo, pBlockData, pLastBlockReader);
// currently loaded file data block is consumed // currently loaded file data block is consumed
if (pBlockData->nRow > 0 && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) { if ((pBlockData->nRow > 0) && (pDumpInfo->rowIndex >= pBlockData->nRow || pDumpInfo->rowIndex < 0)) {
SBlock* pBlock = getCurrentBlock(&pReader->status.blockIter); SBlock* pBlock = getCurrentBlock(&pReader->status.blockIter);
setBlockAllDumped(pDumpInfo, pBlock, pReader->order); setBlockAllDumped(pDumpInfo, pBlock, pReader->order);
break; break;
@ -2252,7 +2259,7 @@ static int32_t doLoadRelatedLastBlock(SLastBlockReader* pLastBlockReader, uint64
} }
static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) { static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status; SReaderStatus* pStatus = &pReader->status;
SLastBlockReader* pLastBlockReader = pStatus->fileIter.pLastBlockReader; SLastBlockReader* pLastBlockReader = pStatus->fileIter.pLastBlockReader;
while(1) { while(1) {
@ -2261,17 +2268,42 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
if (pStatus->pTableIter == NULL) { if (pStatus->pTableIter == NULL) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
} else { // let's try next table }
pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
if (pStatus->pTableIter == NULL) { // load the last data block of current table
return TSDB_CODE_SUCCESS; // todo opt perf by avoiding load last block repeatly
STableBlockScanInfo* pScanInfo = pStatus->pTableIter;
int32_t code = doLoadRelatedLastBlock(pLastBlockReader, pScanInfo->uid, pReader);
if (code != TSDB_CODE_SUCCESS) { // todo handle error
return code;
}
initLastBlockReader(pLastBlockReader, pScanInfo->uid, &pReader->window, &pReader->verRange, &pScanInfo->indexInBlockL);
if (pScanInfo->indexInBlockL == -1) {
bool hasData = nextRowInLastBlock(pLastBlockReader);
if (!hasData) { // current table does not have rows in last block, try next table
pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
if (pStatus->pTableIter == NULL) {
return TSDB_CODE_SUCCESS;
}
continue;
} }
} }
// find the last block that contain the specified block uid code = doBuildDataBlock(pReader);
return doLoadRelatedLastBlock(pLastBlockReader, pStatus->pTableIter->uid, pReader); if (code != TSDB_CODE_SUCCESS) {
return code;
}
//todo check for all empty table if (pReader->pResBlock->info.rows > 0) {
return TSDB_CODE_SUCCESS;
}
// current table is exhausted, let's try next table
pStatus->pTableIter = taosHashIterate(pStatus->pTableMap, pStatus->pTableIter);
if (pStatus->pTableIter == NULL) {
return TSDB_CODE_SUCCESS;
}
} }
} }
@ -2281,8 +2313,8 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
SReaderStatus* pStatus = &pReader->status; SReaderStatus* pStatus = &pReader->status;
SDataBlockIter* pBlockIter = &pStatus->blockIter; SDataBlockIter* pBlockIter = &pStatus->blockIter;
SBlock* pBlock = NULL;
TSDBKEY key = {0}; TSDBKEY key = {0};
SBlock* pBlock = NULL;
STableBlockScanInfo* pScanInfo = NULL; STableBlockScanInfo* pScanInfo = NULL;
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter); SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader; SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader;
@ -2292,29 +2324,28 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
pBlock = getCurrentBlock(pBlockIter); pBlock = getCurrentBlock(pBlockIter);
key = getCurrentKeyInBuf(pBlockIter, pReader); key = getCurrentKeyInBuf(pBlockIter, pReader);
// load the last data block of current table
code = doLoadRelatedLastBlock(pLastBlockReader, pScanInfo->uid, pReader); code = doLoadRelatedLastBlock(pLastBlockReader, pScanInfo->uid, pReader);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
// todo handle error // todo handle error
} }
initLastBlockReader(pLastBlockReader, pBlockInfo->uid, &pReader->window, &pReader->verRange, pScanInfo->indexInBlockL); initLastBlockReader(pLastBlockReader, pBlockInfo->uid, &pReader->window, &pReader->verRange, &pScanInfo->indexInBlockL);
bool hasData = nextRowInLastBlock(pLastBlockReader);
} else {
ASSERT(pBlockIter->numOfBlocks == 0);
} }
if (pBlockInfo == NULL || fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, key, pLastBlockReader)) { if (pBlockInfo == NULL) { // build data block from last data file
if (pBlockInfo != NULL) { ASSERT(pBlockIter->numOfBlocks == 0);
tBlockDataReset(&pStatus->fileBlockData); code = buildComposedDataBlock(pReader);
code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pScanInfo->uid, pReader->pSchema); } else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, key, pLastBlockReader)) {
if (code != TSDB_CODE_SUCCESS) { tBlockDataReset(&pStatus->fileBlockData);
//todo code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pScanInfo->uid, pReader->pSchema);
} if (code != TSDB_CODE_SUCCESS) {
// todo
}
code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData); code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
}
} }
// build composed data block // build composed data block
@ -2425,7 +2456,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
return code; return code;
} }
// all data blocks are check in last file, now let's try the next file // all data blocks are checked in this last block file, now let's try the next file
if (pReader->status.pTableIter == NULL) { if (pReader->status.pTableIter == NULL) {
code = initForFirstBlockInFile(pReader, pBlockIter); code = initForFirstBlockInFile(pReader, pBlockIter);
@ -2434,7 +2465,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
return code; return code;
} }
// this file does not have blocks, let's start check the last block file // this file does not have data files, let's start check the last block file if exists
if (pBlockIter->numOfBlocks == 0) { if (pBlockIter->numOfBlocks == 0) {
goto _begin; goto _begin;
} }
@ -2855,7 +2886,7 @@ int32_t doMergeRowsInLastBlock(SLastBlockReader* pLastBlockReader, int64_t ts, S
while(nextRowInLastBlock(pLastBlockReader)) { while(nextRowInLastBlock(pLastBlockReader)) {
int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader); int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
if (next1 == ts) { if (next1 == ts) {
TSDBROW fRow1 = tsdbRowFromBlockData(&pLastBlockReader->lastBlockData, pLastBlockReader->rowIndex); TSDBROW fRow1 = tsdbRowFromBlockData(&pLastBlockReader->lastBlockData, *pLastBlockReader->rowIndex);
tRowMerge(pMerger, &fRow1); tRowMerge(pMerger, &fRow1);
} else { } else {
break; break;