|
|
|
@ -215,6 +215,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader);
|
|
|
|
|
static TSDBKEY getCurrentKeyInBuf(STableBlockScanInfo* pScanInfo, STsdbReader* pReader);
|
|
|
|
|
static bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo);
|
|
|
|
|
static void initBlockDumpInfo(STsdbReader* pReader, SDataBlockIter* pBlockIter);
|
|
|
|
|
static int32_t getInitialDelIndex(const SArray* pDelSkyline, int32_t order);
|
|
|
|
|
|
|
|
|
|
static bool outOfTimeWindow(int64_t ts, STimeWindow* pWindow) { return (ts > pWindow->ekey) || (ts < pWindow->skey); }
|
|
|
|
|
|
|
|
|
@ -1112,9 +1113,9 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
|
|
|
|
|
|
|
|
|
|
int32_t unDumpedRows = asc ? pBlock->nRow - pDumpInfo->rowIndex : pDumpInfo->rowIndex + 1;
|
|
|
|
|
tsdbDebug("%p copy file block to sdatablock, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
|
|
|
|
|
", rows:%d, remain:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%.2f ms, %s",
|
|
|
|
|
", rows:%d, remain:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", uid:%"PRIu64" elapsed time:%.2f ms, %s",
|
|
|
|
|
pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, dumpedRows,
|
|
|
|
|
unDumpedRows, pBlock->minVer, pBlock->maxVer, elapsedTime, pReader->idStr);
|
|
|
|
|
unDumpedRows, pBlock->minVer, pBlock->maxVer, pBlockInfo->uid, elapsedTime, pReader->idStr);
|
|
|
|
|
|
|
|
|
|
return TSDB_CODE_SUCCESS;
|
|
|
|
|
}
|
|
|
|
@ -2186,17 +2187,17 @@ static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbRea
|
|
|
|
|
if (code == TSDB_CODE_SUCCESS) {
|
|
|
|
|
pBlockScanInfo->iter.hasVal = (tsdbTbDataIterGet(pBlockScanInfo->iter.iter) != NULL);
|
|
|
|
|
|
|
|
|
|
tsdbDebug("%p uid:%" PRId64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
|
|
|
|
|
tsdbDebug("%p uid:%" PRIu64 ", check data in mem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
|
|
|
|
|
"-%" PRId64 " %s",
|
|
|
|
|
pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, d->minKey, d->maxKey, pReader->idStr);
|
|
|
|
|
} else {
|
|
|
|
|
tsdbError("%p uid:%" PRId64 ", failed to create iterator for imem, code:%s, %s", pReader, pBlockScanInfo->uid,
|
|
|
|
|
tsdbError("%p uid:%" PRIu64 ", failed to create iterator for imem, code:%s, %s", pReader, pBlockScanInfo->uid,
|
|
|
|
|
tstrerror(code), pReader->idStr);
|
|
|
|
|
return code;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
tsdbDebug("%p uid:%" PRId64 ", no data in mem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
|
|
|
|
|
tsdbDebug("%p uid:%" PRIu64 ", no data in mem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
STbData* di = NULL;
|
|
|
|
@ -2207,17 +2208,17 @@ static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbRea
|
|
|
|
|
if (code == TSDB_CODE_SUCCESS) {
|
|
|
|
|
pBlockScanInfo->iiter.hasVal = (tsdbTbDataIterGet(pBlockScanInfo->iiter.iter) != NULL);
|
|
|
|
|
|
|
|
|
|
tsdbDebug("%p uid:%" PRId64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
|
|
|
|
|
tsdbDebug("%p uid:%" PRIu64 ", check data in imem from skey:%" PRId64 ", order:%d, ts range in buf:%" PRId64
|
|
|
|
|
"-%" PRId64 " %s",
|
|
|
|
|
pReader, pBlockScanInfo->uid, startKey.ts, pReader->order, di->minKey, di->maxKey, pReader->idStr);
|
|
|
|
|
} else {
|
|
|
|
|
tsdbError("%p uid:%" PRId64 ", failed to create iterator for mem, code:%s, %s", pReader, pBlockScanInfo->uid,
|
|
|
|
|
tsdbError("%p uid:%" PRIu64 ", failed to create iterator for mem, code:%s, %s", pReader, pBlockScanInfo->uid,
|
|
|
|
|
tstrerror(code), pReader->idStr);
|
|
|
|
|
return code;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
} else {
|
|
|
|
|
tsdbDebug("%p uid:%" PRId64 ", no data in imem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
|
|
|
|
|
tsdbDebug("%p uid:%" PRIu64 ", no data in imem, %s", pReader, pBlockScanInfo->uid, pReader->idStr);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
initDelSkylineIterator(pBlockScanInfo, pReader, d, di);
|
|
|
|
@ -2526,6 +2527,14 @@ _end:
|
|
|
|
|
|
|
|
|
|
void setComposedBlockFlag(STsdbReader* pReader, bool composed) { pReader->status.composedDataBlock = composed; }
|
|
|
|
|
|
|
|
|
|
int32_t getInitialDelIndex(const SArray* pDelSkyline, int32_t order) {
|
|
|
|
|
if (pDelSkyline == NULL) {
|
|
|
|
|
return 0;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return ASCENDING_TRAVERSE(order) ? 0 : taosArrayGetSize(pDelSkyline) - 1;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader, STbData* pMemTbData,
|
|
|
|
|
STbData* piMemTbData) {
|
|
|
|
|
if (pBlockScanInfo->delSkyline != NULL) {
|
|
|
|
@ -2543,7 +2552,6 @@ int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader*
|
|
|
|
|
if (pIdx != NULL) {
|
|
|
|
|
code = tsdbReadDelData(pReader->pDelFReader, pIdx, pDelData);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
|
|
|
goto _err;
|
|
|
|
|
}
|
|
|
|
@ -2572,11 +2580,13 @@ int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader*
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
taosArrayDestroy(pDelData);
|
|
|
|
|
pBlockScanInfo->iter.index =
|
|
|
|
|
ASCENDING_TRAVERSE(pReader->order) ? 0 : taosArrayGetSize(pBlockScanInfo->delSkyline) - 1;
|
|
|
|
|
pBlockScanInfo->iiter.index = pBlockScanInfo->iter.index;
|
|
|
|
|
pBlockScanInfo->fileDelIndex = pBlockScanInfo->iter.index;
|
|
|
|
|
pBlockScanInfo->lastBlockDelIndex = pBlockScanInfo->iter.index;
|
|
|
|
|
int32_t index = getInitialDelIndex(pBlockScanInfo->delSkyline, pReader->order);
|
|
|
|
|
|
|
|
|
|
pBlockScanInfo->iter.index = index;
|
|
|
|
|
pBlockScanInfo->iiter.index = index;
|
|
|
|
|
pBlockScanInfo->fileDelIndex = index;
|
|
|
|
|
pBlockScanInfo->lastBlockDelIndex = index;
|
|
|
|
|
|
|
|
|
|
return code;
|
|
|
|
|
|
|
|
|
|
_err:
|
|
|
|
@ -2676,7 +2686,7 @@ static int32_t uidComparFunc(const void* p1, const void* p2) {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
static void extractOrderedTableUidList(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus) {
|
|
|
|
|
static void extractOrderedTableUidList(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus, int32_t order) {
|
|
|
|
|
int32_t index = 0;
|
|
|
|
|
int32_t total = taosHashGetSize(pStatus->pTableMap);
|
|
|
|
|
|
|
|
|
@ -2690,7 +2700,21 @@ static void extractOrderedTableUidList(SUidOrderCheckInfo* pOrderCheckInfo, SRea
|
|
|
|
|
taosSort(pOrderCheckInfo->tableUidList, total, sizeof(uint64_t), uidComparFunc);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
static int32_t initOrderCheckInfo(SUidOrderCheckInfo* pOrderCheckInfo, SReaderStatus* pStatus) {
|
|
|
|
|
// reset the last del file index
|
|
|
|
|
static void resetScanBlockLastBlockDelIndex(SReaderStatus* pStatus, int32_t order) {
|
|
|
|
|
void* p = taosHashIterate(pStatus->pTableMap, NULL);
|
|
|
|
|
while (p != NULL) {
|
|
|
|
|
STableBlockScanInfo* pScanInfo = *(STableBlockScanInfo**)p;
|
|
|
|
|
|
|
|
|
|
// reset the last del file index
|
|
|
|
|
pScanInfo->lastBlockDelIndex = getInitialDelIndex(pScanInfo->delSkyline, order);
|
|
|
|
|
p = taosHashIterate(pStatus->pTableMap, p);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
static int32_t initOrderCheckInfo(SUidOrderCheckInfo* pOrderCheckInfo, STsdbReader* pReader) {
|
|
|
|
|
SReaderStatus* pStatus = &pReader->status;
|
|
|
|
|
|
|
|
|
|
int32_t total = taosHashGetSize(pStatus->pTableMap);
|
|
|
|
|
if (total == 0) {
|
|
|
|
|
return TSDB_CODE_SUCCESS;
|
|
|
|
@ -2703,7 +2727,7 @@ static int32_t initOrderCheckInfo(SUidOrderCheckInfo* pOrderCheckInfo, SReaderSt
|
|
|
|
|
return TSDB_CODE_OUT_OF_MEMORY;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
extractOrderedTableUidList(pOrderCheckInfo, pStatus);
|
|
|
|
|
extractOrderedTableUidList(pOrderCheckInfo, pStatus, pReader->order);
|
|
|
|
|
uint64_t uid = pOrderCheckInfo->tableUidList[0];
|
|
|
|
|
pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
|
|
|
|
|
} else {
|
|
|
|
@ -2720,7 +2744,7 @@ static int32_t initOrderCheckInfo(SUidOrderCheckInfo* pOrderCheckInfo, SReaderSt
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
pOrderCheckInfo->tableUidList = p;
|
|
|
|
|
extractOrderedTableUidList(pOrderCheckInfo, pStatus);
|
|
|
|
|
extractOrderedTableUidList(pOrderCheckInfo, pStatus, pReader->order);
|
|
|
|
|
|
|
|
|
|
uid = pOrderCheckInfo->tableUidList[0];
|
|
|
|
|
pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
|
|
|
|
@ -2740,11 +2764,7 @@ static bool moveToNextTable(SUidOrderCheckInfo* pOrderedCheckInfo, SReaderStatus
|
|
|
|
|
|
|
|
|
|
uint64_t uid = pOrderedCheckInfo->tableUidList[pOrderedCheckInfo->currentIndex];
|
|
|
|
|
pStatus->pTableIter = taosHashGet(pStatus->pTableMap, &uid, sizeof(uid));
|
|
|
|
|
if (pStatus->pTableIter == NULL) {
|
|
|
|
|
return false;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return true;
|
|
|
|
|
return (pStatus->pTableIter != NULL);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
|
|
|
|
@ -2752,7 +2772,7 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
|
|
|
|
|
SLastBlockReader* pLastBlockReader = pStatus->fileIter.pLastBlockReader;
|
|
|
|
|
|
|
|
|
|
SUidOrderCheckInfo* pOrderedCheckInfo = &pStatus->uidCheckInfo;
|
|
|
|
|
int32_t code = initOrderCheckInfo(pOrderedCheckInfo, pStatus);
|
|
|
|
|
int32_t code = initOrderCheckInfo(pOrderedCheckInfo, pReader);
|
|
|
|
|
if (code != TSDB_CODE_SUCCESS || (taosHashGetSize(pStatus->pTableMap) == 0)) {
|
|
|
|
|
return code;
|
|
|
|
|
}
|
|
|
|
@ -2817,6 +2837,8 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
|
|
|
|
|
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter);
|
|
|
|
|
SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader;
|
|
|
|
|
|
|
|
|
|
ASSERT(pBlockInfo != NULL);
|
|
|
|
|
|
|
|
|
|
if (pBlockInfo != NULL) {
|
|
|
|
|
pScanInfo =
|
|
|
|
|
*(STableBlockScanInfo**)taosHashGet(pReader->status.pTableMap, &pBlockInfo->uid, sizeof(pBlockInfo->uid));
|
|
|
|
@ -2837,7 +2859,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
|
|
|
|
|
initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
|
|
|
|
|
TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader);
|
|
|
|
|
|
|
|
|
|
if (pBlockInfo == NULL) { // build data block from last data file
|
|
|
|
|
/*if (pBlockInfo == NULL) { // build data block from last data file
|
|
|
|
|
SBlockData* pBData = &pReader->status.fileBlockData;
|
|
|
|
|
tBlockDataReset(pBData);
|
|
|
|
|
|
|
|
|
@ -2869,7 +2891,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
|
|
|
|
|
pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
|
|
|
|
|
pResBlock->info.rows, el, pReader->idStr);
|
|
|
|
|
}
|
|
|
|
|
} else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) {
|
|
|
|
|
} else*/ if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) {
|
|
|
|
|
code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData, pScanInfo->uid);
|
|
|
|
|
if (code != TSDB_CODE_SUCCESS) {
|
|
|
|
|
return code;
|
|
|
|
@ -3040,6 +3062,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
|
|
|
|
|
|
|
|
|
|
// this file does not have data files, let's start check the last block file if exists
|
|
|
|
|
if (pBlockIter->numOfBlocks == 0) {
|
|
|
|
|
resetScanBlockLastBlockDelIndex(&pReader->status, pReader->order);
|
|
|
|
|
goto _begin;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
@ -3071,6 +3094,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
|
|
|
|
|
// data blocks in current file are exhausted, let's try the next file now
|
|
|
|
|
tBlockDataReset(&pReader->status.fileBlockData);
|
|
|
|
|
resetDataBlockIterator(pBlockIter, pReader->order);
|
|
|
|
|
resetScanBlockLastBlockDelIndex(&pReader->status, pReader->order);
|
|
|
|
|
goto _begin;
|
|
|
|
|
} else {
|
|
|
|
|
code = initForFirstBlockInFile(pReader, pBlockIter);
|
|
|
|
@ -3082,6 +3106,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
|
|
|
|
|
|
|
|
|
|
// this file does not have blocks, let's start check the last block file
|
|
|
|
|
if (pBlockIter->numOfBlocks == 0) {
|
|
|
|
|
resetScanBlockLastBlockDelIndex(&pReader->status, pReader->order);
|
|
|
|
|
goto _begin;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|