fix(query): fix the invalid the check of last file blocks.
This commit is contained in:
parent
e66583be20
commit
a6cece10df
|
@ -633,7 +633,7 @@ typedef struct SMergeTree {
|
|||
struct SLDataIter *pIter;
|
||||
} SMergeTree;
|
||||
|
||||
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader, uint64_t uid, STimeWindow* pTimeWindow, SVersionRange* pVerRange);
|
||||
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader, uint64_t suid, uint64_t uid, STimeWindow* pTimeWindow, SVersionRange* pVerRange);
|
||||
void tMergeTreeAddIter(SMergeTree *pMTree, struct SLDataIter *pIter);
|
||||
bool tMergeTreeNext(SMergeTree *pMTree);
|
||||
TSDBROW tMergeTreeGetRow(SMergeTree *pMTree);
|
||||
|
|
|
@ -420,6 +420,7 @@ typedef enum {
|
|||
typedef struct {
|
||||
SFSLASTNEXTROWSTATES state; // [input]
|
||||
STsdb *pTsdb; // [input]
|
||||
tb_uid_t suid;
|
||||
tb_uid_t uid;
|
||||
int32_t nFileSet;
|
||||
int32_t iFileSet;
|
||||
|
@ -454,7 +455,7 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
|
|||
code = tsdbDataFReaderOpen(&state->pDataFReader, state->pTsdb, pFileSet);
|
||||
if (code) goto _err;
|
||||
|
||||
tMergeTreeOpen(&state->mergeTree, 1, state->pDataFReader, state->uid,
|
||||
tMergeTreeOpen(&state->mergeTree, 1, state->pDataFReader, state->suid, state->uid,
|
||||
&(STimeWindow){.skey = TSKEY_MIN, .ekey = TSKEY_MAX},
|
||||
&(SVersionRange){.minVer = 0, .maxVer = UINT64_MAX});
|
||||
bool hasVal = tMergeTreeNext(&state->mergeTree);
|
||||
|
@ -796,7 +797,7 @@ static bool tsdbKeyDeleted(TSDBKEY *key, SArray *pSkyline, int64_t *iSkyline) {
|
|||
if (key->ts > pItemBack->ts) {
|
||||
return false;
|
||||
} else if (key->ts >= pItemFront->ts && key->ts <= pItemBack->ts) {
|
||||
if ((key->version <= pItemFront->version || key->ts == pItemBack->ts && key->version <= pItemBack->version)) {
|
||||
if (key->version <= pItemFront->version || (key->ts == pItemBack->ts && key->version <= pItemBack->version)) {
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
|
@ -890,6 +891,7 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
|
|||
pIter->fsLastState.state = (SFSLASTNEXTROWSTATES)SFSNEXTROW_FS;
|
||||
pIter->fsLastState.pTsdb = pTsdb;
|
||||
pIter->fsLastState.aDFileSet = pIter->pReadSnap->fs.aDFileSet;
|
||||
pIter->fsLastState.suid = suid;
|
||||
pIter->fsLastState.uid = uid;
|
||||
|
||||
pIter->fsState.state = SFSNEXTROW_FS;
|
||||
|
|
|
@ -42,8 +42,8 @@ static SBlockData* getNextBlock(SLDataIter* pIter) {
|
|||
return getCurrentBlock(pIter);
|
||||
}
|
||||
|
||||
int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t iSst, int8_t backward, uint64_t uid,
|
||||
STimeWindow *pTimeWindow, SVersionRange *pRange) {
|
||||
int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t iSst, int8_t backward, uint64_t suid,
|
||||
uint64_t uid, STimeWindow *pTimeWindow, SVersionRange *pRange) {
|
||||
int32_t code = 0;
|
||||
*pIter = taosMemoryCalloc(1, sizeof(SLDataIter));
|
||||
if (*pIter == NULL) {
|
||||
|
@ -52,11 +52,11 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
|
|||
}
|
||||
|
||||
(*pIter)->uid = uid;
|
||||
(*pIter)->timeWindow = *pTimeWindow;
|
||||
(*pIter)->verRange = *pRange;
|
||||
(*pIter)->pReader = pReader;
|
||||
(*pIter)->iSst = iSst;
|
||||
(*pIter)->pReader = pReader;
|
||||
(*pIter)->verRange = *pRange;
|
||||
(*pIter)->backward = backward;
|
||||
(*pIter)->timeWindow = *pTimeWindow;
|
||||
(*pIter)->aSstBlk = taosArrayInit(0, sizeof(SSstBlk));
|
||||
if ((*pIter)->aSstBlk == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
|
@ -85,6 +85,10 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
|
|||
if (!backward) { // asc
|
||||
for (int32_t i = 0; i < size; ++i) {
|
||||
SSstBlk *p = taosArrayGet((*pIter)->aSstBlk, i);
|
||||
if (p->suid != suid) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (p->minUid <= uid && p->maxUid >= uid) {
|
||||
index = i;
|
||||
break;
|
||||
|
@ -93,6 +97,10 @@ int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t
|
|||
} else { // desc
|
||||
for (int32_t i = size - 1; i >= 0; --i) {
|
||||
SSstBlk *p = taosArrayGet((*pIter)->aSstBlk, i);
|
||||
if (p->suid != suid) {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (p->minUid <= uid && p->maxUid >= uid) {
|
||||
index = i;
|
||||
break;
|
||||
|
@ -134,9 +142,31 @@ void tLDataIterNextBlock(SLDataIter *pIter) {
|
|||
break;
|
||||
}
|
||||
|
||||
// check uid firstly
|
||||
if (p->minUid <= pIter->uid && p->maxUid >= pIter->uid) {
|
||||
index = i;
|
||||
break;
|
||||
if ((!pIter->backward) && p->minKey > pIter->timeWindow.ekey) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (pIter->backward && p->maxKey < pIter->timeWindow.skey) {
|
||||
break;
|
||||
}
|
||||
|
||||
// check time range secondly
|
||||
if (p->minKey <= pIter->timeWindow.ekey && p->maxKey >= pIter->timeWindow.skey) {
|
||||
if ((!pIter->backward) && p->minVer > pIter->verRange.maxVer) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (pIter->backward && p->maxVer < pIter->verRange.minVer) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (p->minVer <= pIter->verRange.maxVer && p->maxVer >= pIter->verRange.minVer) {
|
||||
index = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -196,14 +226,6 @@ static void findNextValidRow(SLDataIter *pIter) {
|
|||
continue;
|
||||
}
|
||||
|
||||
// todo handle delete soon
|
||||
#if 0
|
||||
TSDBKEY k = {.ts = ts, .version = ver};
|
||||
if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order)) {
|
||||
continue;
|
||||
}
|
||||
#endif
|
||||
|
||||
hasVal = true;
|
||||
break;
|
||||
}
|
||||
|
@ -294,7 +316,7 @@ static FORCE_INLINE int32_t tLDataIterCmprFn(const void *p1, const void *p2) {
|
|||
}
|
||||
}
|
||||
|
||||
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t uid,
|
||||
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
|
||||
STimeWindow *pTimeWindow, SVersionRange *pVerRange) {
|
||||
pMTree->backward = backward;
|
||||
pMTree->pIter = NULL;
|
||||
|
@ -308,7 +330,7 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead
|
|||
|
||||
struct SLDataIter *pIterList[TSDB_DEFAULT_LAST_FILE] = {0};
|
||||
for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) { // open all last file
|
||||
code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange);
|
||||
code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, suid, uid, pTimeWindow, pVerRange);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
goto _end;
|
||||
}
|
||||
|
|
|
@ -1934,7 +1934,7 @@ static bool initLastBlockReader(SLastBlockReader* pLastBlockReader, STableBlockS
|
|||
pLastBlockReader->uid = pBlockScanInfo->uid;
|
||||
int32_t code =
|
||||
tMergeTreeOpen(&pLastBlockReader->mergeTree, (pLastBlockReader->order == TSDB_ORDER_DESC), pReader->pFileReader,
|
||||
pBlockScanInfo->uid, &pLastBlockReader->window, &pLastBlockReader->verRange);
|
||||
pReader->suid, pBlockScanInfo->uid, &pLastBlockReader->window, &pLastBlockReader->verRange);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return false;
|
||||
}
|
||||
|
@ -2537,22 +2537,24 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
|
|||
bool hasNext = blockIteratorNext(&pReader->status.blockIter);
|
||||
if (hasNext) { // check for the next block in the block accessed order list
|
||||
initBlockDumpInfo(pReader, pBlockIter);
|
||||
} else if (hasDataInLastBlock(pReader->status.fileIter.pLastBlockReader)) {
|
||||
// data blocks in current file are exhausted, let's try the next file now
|
||||
tBlockDataReset(&pReader->status.fileBlockData);
|
||||
resetDataBlockIterator(pBlockIter, pReader->order);
|
||||
goto _begin;
|
||||
} else {
|
||||
code = initForFirstBlockInFile(pReader, pBlockIter);
|
||||
|
||||
// error happens or all the data files are completely checked
|
||||
if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
|
||||
return code;
|
||||
}
|
||||
|
||||
// this file does not have blocks, let's start check the last block file
|
||||
if (pBlockIter->numOfBlocks == 0) {
|
||||
if (pReader->status.pCurrentFileset->nSstF > 0) {
|
||||
// data blocks in current file are exhausted, let's try the next file now
|
||||
tBlockDataReset(&pReader->status.fileBlockData);
|
||||
resetDataBlockIterator(pBlockIter, pReader->order);
|
||||
goto _begin;
|
||||
} else {
|
||||
code = initForFirstBlockInFile(pReader, pBlockIter);
|
||||
|
||||
// error happens or all the data files are completely checked
|
||||
if ((code != TSDB_CODE_SUCCESS) || (pReader->status.loadFromFile == false)) {
|
||||
return code;
|
||||
}
|
||||
|
||||
// this file does not have blocks, let's start check the last block file
|
||||
if (pBlockIter->numOfBlocks == 0) {
|
||||
goto _begin;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue