fix(tsdb): enable pin stt datablock.
This commit is contained in:
parent
f90cf54714
commit
f751f2bb3d
|
@ -714,13 +714,18 @@ typedef struct SSttBlockLoadCostInfo {
|
|||
double statisElapsedTime;
|
||||
} SSttBlockLoadCostInfo;
|
||||
|
||||
typedef struct SBlockDataInfo {
|
||||
SBlockData data;
|
||||
bool pin;
|
||||
int32_t sttBlockIndex;
|
||||
} SBlockDataInfo;
|
||||
|
||||
typedef struct SSttBlockLoadInfo {
|
||||
SBlockData blockData[2]; // buffered block data
|
||||
SBlockDataInfo blockData[2]; // buffered block data
|
||||
int32_t statisBlockIndex; // buffered statistics block index
|
||||
void *statisBlock; // buffered statistics block data
|
||||
void *pSttStatisBlkArray;
|
||||
SArray *aSttBlk;
|
||||
int32_t blockIndex[2]; // to denote the loaded block in the corresponding position.
|
||||
int32_t currentLoadBlockIndex;
|
||||
STSchema *pSchema;
|
||||
int16_t *colIds;
|
||||
|
@ -797,9 +802,6 @@ struct SLDataIter {
|
|||
};
|
||||
|
||||
#define tMergeTreeGetRow(_t) (&((_t)->pIter->rInfo.row))
|
||||
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
|
||||
STimeWindow *pTimeWindow, SVersionRange *pVerRange, SSttBlockLoadInfo *pBlockLoadInfo,
|
||||
bool destroyLoadInfo, const char *idStr, bool strictTimeRange, SLDataIter *pLDataIter);
|
||||
|
||||
struct SSttFileReader;
|
||||
typedef int32_t (*_load_tomb_fn)(STsdbReader *pReader, struct SSttFileReader *pSttFileReader,
|
||||
|
@ -822,10 +824,13 @@ typedef struct {
|
|||
void *pReader;
|
||||
void *idstr;
|
||||
} SMergeTreeConf;
|
||||
|
||||
int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf);
|
||||
|
||||
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
|
||||
bool tMergeTreeNext(SMergeTree *pMTree);
|
||||
void tMergeTreePinSttBlock(SMergeTree* pMTree);
|
||||
void tMergeTreeUnpinSttBlock(SMergeTree* pMTree);
|
||||
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);
|
||||
void tMergeTreeClose(SMergeTree *pMTree);
|
||||
|
||||
|
|
|
@ -29,16 +29,17 @@ SSttBlockLoadInfo *tCreateOneLastBlockLoadInfo(STSchema *pSchema, int16_t *colLi
|
|||
return NULL;
|
||||
}
|
||||
|
||||
pLoadInfo->blockIndex[0] = -1;
|
||||
pLoadInfo->blockIndex[1] = -1;
|
||||
pLoadInfo->blockData[0].sttBlockIndex = -1;
|
||||
pLoadInfo->blockData[1].sttBlockIndex = -1;
|
||||
|
||||
pLoadInfo->currentLoadBlockIndex = 1;
|
||||
|
||||
int32_t code = tBlockDataCreate(&pLoadInfo->blockData[0]);
|
||||
int32_t code = tBlockDataCreate(&pLoadInfo->blockData[0].data);
|
||||
if (code) {
|
||||
terrno = code;
|
||||
}
|
||||
|
||||
code = tBlockDataCreate(&pLoadInfo->blockData[1]);
|
||||
code = tBlockDataCreate(&pLoadInfo->blockData[1].data);
|
||||
if (code) {
|
||||
terrno = code;
|
||||
}
|
||||
|
@ -66,11 +67,16 @@ void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo) {
|
|||
}
|
||||
|
||||
pLoadInfo->currentLoadBlockIndex = 1;
|
||||
pLoadInfo->blockIndex[0] = -1;
|
||||
pLoadInfo->blockIndex[1] = -1;
|
||||
|
||||
tBlockDataDestroy(&pLoadInfo->blockData[0]);
|
||||
tBlockDataDestroy(&pLoadInfo->blockData[1]);
|
||||
SBlockDataInfo* pInfo = &pLoadInfo->blockData[0];
|
||||
tBlockDataDestroy(&pInfo->data);
|
||||
pInfo->sttBlockIndex = -1;
|
||||
pInfo->pin = false;
|
||||
|
||||
pInfo = &pLoadInfo->blockData[1];
|
||||
tBlockDataDestroy(&pInfo->data);
|
||||
pInfo->sttBlockIndex = -1;
|
||||
pInfo->pin = false;
|
||||
|
||||
taosArrayDestroy(pLoadInfo->aSttBlk);
|
||||
taosMemoryFree(pLoadInfo);
|
||||
|
@ -109,39 +115,48 @@ void *destroySttBlockReader(SArray *pLDataIterArray, SSttBlockLoadCostInfo* pLoa
|
|||
return NULL;
|
||||
}
|
||||
|
||||
// choose the unpinned slot to load next data block
|
||||
static void updateBlockLoadSlot(SSttBlockLoadInfo* pLoadInfo) {
|
||||
int32_t nextSlotIndex = pLoadInfo->currentLoadBlockIndex ^ 1;
|
||||
if (pLoadInfo->blockData[nextSlotIndex].pin) {
|
||||
nextSlotIndex = nextSlotIndex ^ 1;
|
||||
}
|
||||
|
||||
pLoadInfo->currentLoadBlockIndex = nextSlotIndex;
|
||||
}
|
||||
|
||||
static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
|
||||
int32_t code = 0;
|
||||
|
||||
SSttBlockLoadInfo *pInfo = pIter->pBlockLoadInfo;
|
||||
if (pInfo->blockIndex[0] == pIter->iSttBlk) {
|
||||
if (pInfo->blockData[0].sttBlockIndex == pIter->iSttBlk) {
|
||||
if (pInfo->currentLoadBlockIndex != 0) {
|
||||
tsdbDebug("current load index is set to 0, block index:%d, fileVer:%" PRId64 ", due to uid:%" PRIu64
|
||||
", load data, %s",
|
||||
pIter->iSttBlk, pIter->cid, pIter->uid, idStr);
|
||||
pInfo->currentLoadBlockIndex = 0;
|
||||
}
|
||||
return &pInfo->blockData[0];
|
||||
return &pInfo->blockData[0].data;
|
||||
}
|
||||
|
||||
if (pInfo->blockIndex[1] == pIter->iSttBlk) {
|
||||
if (pInfo->blockData[1].sttBlockIndex == pIter->iSttBlk) {
|
||||
if (pInfo->currentLoadBlockIndex != 1) {
|
||||
tsdbDebug("current load index is set to 1, block index:%d, fileVer:%" PRId64 ", due to uid:%" PRIu64
|
||||
", load data, %s",
|
||||
pIter->iSttBlk, pIter->cid, pIter->uid, idStr);
|
||||
pInfo->currentLoadBlockIndex = 1;
|
||||
}
|
||||
return &pInfo->blockData[1];
|
||||
return &pInfo->blockData[1].data;
|
||||
}
|
||||
|
||||
if (pIter->pSttBlk == NULL || pInfo->pSchema == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
// current block not loaded yet
|
||||
pInfo->currentLoadBlockIndex ^= 1;
|
||||
updateBlockLoadSlot(pInfo);
|
||||
int64_t st = taosGetTimestampUs();
|
||||
|
||||
SBlockData *pBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex];
|
||||
SBlockData *pBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex].data;
|
||||
code = tsdbSttFileReadBlockDataByColumn(pIter->pReader, pIter->pSttBlk, pBlock, pInfo->pSchema, &pInfo->colIds[1],
|
||||
pInfo->numOfCols - 1);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
|
@ -159,12 +174,12 @@ static SBlockData *loadLastBlock(SLDataIter *pIter, const char *idStr) {
|
|||
pIter->pSttBlk->minUid, pIter->pSttBlk->maxUid, pIter->pSttBlk->minKey, pIter->pSttBlk->maxKey, pBlock, el,
|
||||
idStr);
|
||||
|
||||
pInfo->blockIndex[pInfo->currentLoadBlockIndex] = pIter->iSttBlk;
|
||||
pIter->iRow = (pIter->backward) ? pInfo->blockData[pInfo->currentLoadBlockIndex].nRow : -1;
|
||||
pInfo->blockData[pInfo->currentLoadBlockIndex].sttBlockIndex = pIter->iSttBlk;
|
||||
pIter->iRow = (pIter->backward) ? pInfo->blockData[pInfo->currentLoadBlockIndex].data.nRow : -1;
|
||||
|
||||
tsdbDebug("last block index list:%d, %d, rowIndex:%d %s", pInfo->blockIndex[0], pInfo->blockIndex[1], pIter->iRow,
|
||||
idStr);
|
||||
return &pInfo->blockData[pInfo->currentLoadBlockIndex];
|
||||
tsdbDebug("last block index list:%d, %d, rowIndex:%d %s", pInfo->blockData[0].sttBlockIndex,
|
||||
pInfo->blockData[1].sttBlockIndex, pIter->iRow, idStr);
|
||||
return &pInfo->blockData[pInfo->currentLoadBlockIndex].data;
|
||||
|
||||
_exit:
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
|
@ -825,8 +840,60 @@ void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) { tRBTreePut(&pMTr
|
|||
|
||||
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree) { return pMTree->ignoreEarlierTs; }
|
||||
|
||||
static void tLDataIterPinSttBlock(SLDataIter* pIter, const char* id) {
|
||||
SSttBlockLoadInfo* pInfo = pIter->pBlockLoadInfo;
|
||||
|
||||
if (pInfo->blockData[0].sttBlockIndex == pIter->iSttBlk) {
|
||||
pInfo->blockData[0].pin = true;
|
||||
ASSERT(!pInfo->blockData[1].pin);
|
||||
return;
|
||||
}
|
||||
|
||||
if (pInfo->blockData[1].sttBlockIndex == pIter->iSttBlk) {
|
||||
pInfo->blockData[1].pin = true;
|
||||
ASSERT(!pInfo->blockData[0].pin);
|
||||
return;
|
||||
}
|
||||
|
||||
tsdbError("failed to pin any stt block, sttBlock:%d", pIter->iSttBlk, id);
|
||||
}
|
||||
|
||||
static void tLDataIterUnpinSttBlock(SLDataIter* pIter, const char* id) {
|
||||
SSttBlockLoadInfo* pInfo = pIter->pBlockLoadInfo;
|
||||
if (pInfo->blockData[0].pin) {
|
||||
ASSERT(!pInfo->blockData[1].pin);
|
||||
pInfo->blockData[0].pin = false;
|
||||
return;
|
||||
}
|
||||
|
||||
if (pInfo->blockData[1].pin) {
|
||||
ASSERT(!pInfo->blockData[0].pin);
|
||||
pInfo->blockData[1].pin = false;
|
||||
return;
|
||||
}
|
||||
|
||||
tsdbError("failed to unpin any stt block, sttBlock:%d", pIter->iSttBlk, id);
|
||||
}
|
||||
|
||||
void tMergeTreePinSttBlock(SMergeTree *pMTree) {
|
||||
if (pMTree->pIter == NULL) {
|
||||
return;
|
||||
}
|
||||
|
||||
SLDataIter* pIter = pMTree->pIter;
|
||||
tLDataIterPinSttBlock(pIter, pMTree->idStr);
|
||||
}
|
||||
|
||||
void tMergeTreeUnpinSttBlock(SMergeTree *pMTree) {
|
||||
if (pMTree->pIter == NULL) {
|
||||
return;
|
||||
}
|
||||
|
||||
SLDataIter* pIter = pMTree->pIter;
|
||||
tLDataIterUnpinSttBlock(pIter, pMTree->idStr);
|
||||
}
|
||||
|
||||
bool tMergeTreeNext(SMergeTree *pMTree) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
if (pMTree->pIter) {
|
||||
SLDataIter *pIter = pMTree->pIter;
|
||||
|
||||
|
|
|
@ -1420,13 +1420,24 @@ static bool nextRowFromLastBlocks(SLastBlockReader* pLastBlockReader, STableBloc
|
|||
}
|
||||
}
|
||||
|
||||
static void doPinSttBlock(SLastBlockReader* pLastBlockReader) {
|
||||
tMergeTreePinSttBlock(&pLastBlockReader->mergeTree);
|
||||
}
|
||||
|
||||
static void doUnpinSttBlock(SLastBlockReader* pLastBlockReader) {
|
||||
tMergeTreeUnpinSttBlock(&pLastBlockReader->mergeTree);
|
||||
}
|
||||
|
||||
static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SLastBlockReader* pLastBlockReader,
|
||||
STableBlockScanInfo* pScanInfo, int64_t ts, STsdbReader* pReader,
|
||||
bool* copied) {
|
||||
int32_t code = TSDB_CODE_SUCCESS;
|
||||
*copied = false;
|
||||
|
||||
// avoid the fetch next row replace the referenced stt block in buffer
|
||||
doPinSttBlock(pLastBlockReader);
|
||||
bool hasVal = nextRowFromLastBlocks(pLastBlockReader, pScanInfo, &pReader->info.verRange);
|
||||
doUnpinSttBlock(pLastBlockReader);
|
||||
if (hasVal) {
|
||||
int64_t next1 = getCurrentKeyInLastBlock(pLastBlockReader);
|
||||
if (next1 != ts) {
|
||||
|
|
|
@ -146,7 +146,6 @@ typedef struct SLastBlockReader {
|
|||
int32_t order;
|
||||
uint64_t uid;
|
||||
SMergeTree mergeTree;
|
||||
SSttBlockLoadInfo* pInfo;
|
||||
int64_t currentKey;
|
||||
} SLastBlockReader;
|
||||
|
||||
|
|
Loading…
Reference in New Issue