Merge branch 'feat/tsdb_refact' of github.com:taosdata/TDengine into feat/tsdb_refact
This commit is contained in:
Cary Xu
commit
9fab97a01e
|
|
@ -35,7 +35,6 @@ extern "C" {
|
|||
typedef struct TSDBROW TSDBROW;
|
||||
typedef struct TABLEID TABLEID;
|
||||
typedef struct TSDBKEY TSDBKEY;
|
||||
typedef struct KEYINFO KEYINFO;
|
||||
typedef struct SDelData SDelData;
|
||||
typedef struct SDelIdx SDelIdx;
|
||||
typedef struct STbData STbData;
|
||||
|
|
@ -43,7 +42,6 @@ typedef struct SMemTable SMemTable;
|
|||
typedef struct STbDataIter STbDataIter;
|
||||
typedef struct STable STable;
|
||||
typedef struct SMapData SMapData;
|
||||
typedef struct SBlockSMA SBlockSMA;
|
||||
typedef struct SBlockIdx SBlockIdx;
|
||||
typedef struct SBlock SBlock;
|
||||
typedef struct SBlockStatis SBlockStatis;
|
||||
|
|
@ -106,14 +104,6 @@ int32_t tTABLEIDCmprFn(const void *p1, const void *p2);
|
|||
int32_t tsdbKeyCmprFn(const void *p1, const void *p2);
|
||||
#define MIN_TSDBKEY(KEY1, KEY2) ((tsdbKeyCmprFn(&(KEY1), &(KEY2)) < 0) ? (KEY1) : (KEY2))
|
||||
#define MAX_TSDBKEY(KEY1, KEY2) ((tsdbKeyCmprFn(&(KEY1), &(KEY2)) > 0) ? (KEY1) : (KEY2))
|
||||
// KEYINFO
|
||||
#define tKEYINFOInit() \
|
||||
((KEYINFO){.maxKey = {.ts = TSKEY_MIN, .version = -1}, \
|
||||
.minKey = {.ts = TSKEY_MAX, .version = INT64_MAX}, \
|
||||
.minVerion = INT64_MAX, \
|
||||
.maxVersion = -1})
|
||||
int32_t tPutKEYINFO(uint8_t *p, KEYINFO *pKeyInfo);
|
||||
int32_t tGetKEYINFO(uint8_t *p, KEYINFO *pKeyInfo);
|
||||
// SBlockCol
|
||||
int32_t tPutBlockCol(uint8_t *p, void *ph);
|
||||
int32_t tGetBlockCol(uint8_t *p, void *ph);
|
||||
|
|
@ -123,6 +113,7 @@ void tBlockReset(SBlock *pBlock);
|
|||
int32_t tPutBlock(uint8_t *p, void *ph);
|
||||
int32_t tGetBlock(uint8_t *p, void *ph);
|
||||
int32_t tBlockCmprFn(const void *p1, const void *p2);
|
||||
bool tBlockHasSma(SBlock *pBlock);
|
||||
// SBlockIdx
|
||||
// #define tBlockIdxInit(SUID, UID) ((SBlockIdx){.suid = (SUID), .uid = (UID), .info = tKEYINFOInit()})
|
||||
void tBlockIdxReset(SBlockIdx *pBlockIdx);
|
||||
|
|
@ -173,6 +164,7 @@ int32_t tGetMapData(uint8_t *p, SMapData *pMapData);
|
|||
int32_t tsdbKeyFid(TSKEY key, int32_t minutes, int8_t precision);
|
||||
void tsdbFidKeyRange(int32_t fid, int32_t minutes, int8_t precision, TSKEY *minKey, TSKEY *maxKey);
|
||||
int32_t tsdbBuildDeleteSkyline(SArray *aDelData, int32_t sidx, int32_t eidx, SArray *aSkyline);
|
||||
void tsdbCalcColDataSMA(SColData *pColData, SColumnDataAgg *pColAgg);
|
||||
// tsdbMemTable ==============================================================================================
|
||||
// SMemTable
|
||||
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
|
||||
|
|
@ -230,7 +222,7 @@ int32_t tsdbReadColData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBl
|
|||
SBlockData *pBlockData, uint8_t **ppBuf1, uint8_t **ppBuf2);
|
||||
int32_t tsdbReadBlockData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBlock, SBlockData *pBlockData,
|
||||
uint8_t **ppBuf1, uint8_t **ppBuf2);
|
||||
int32_t tsdbReadBlockSMA(SDataFReader *pReader, SBlockSMA *pBlkSMA);
|
||||
int32_t tsdbReadBlockSma(SDataFReader *pReader, SBlock *pBlock, SArray *aColumnDataAgg, uint8_t **ppBuf);
|
||||
// SDelFWriter
|
||||
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb);
|
||||
int32_t tsdbDelFWriterClose(SDelFWriter *pWriter, int8_t sync);
|
||||
|
|
@ -287,13 +279,6 @@ struct TSDBKEY {
|
|||
TSKEY ts;
|
||||
};
|
||||
|
||||
struct KEYINFO {
|
||||
TSDBKEY minKey;
|
||||
TSDBKEY maxKey;
|
||||
int64_t minVerion;
|
||||
int64_t maxVersion;
|
||||
};
|
||||
|
||||
typedef struct SMemSkipListNode SMemSkipListNode;
|
||||
struct SMemSkipListNode {
|
||||
int8_t level;
|
||||
|
|
@ -477,21 +462,6 @@ struct SDelFile {
|
|||
int64_t offset;
|
||||
};
|
||||
|
||||
typedef struct {
|
||||
int16_t colId;
|
||||
int16_t maxIndex;
|
||||
int16_t minIndex;
|
||||
int16_t numOfNull;
|
||||
int64_t sum;
|
||||
int64_t max;
|
||||
int64_t min;
|
||||
} SColSMA;
|
||||
|
||||
struct SBlockSMA {
|
||||
int32_t nCol;
|
||||
SColSMA *aColSMA;
|
||||
};
|
||||
|
||||
#pragma pack(push, 1)
|
||||
struct SBlockDataHdr {
|
||||
uint32_t delimiter;
|
||||
|
|
|
|||
|
|
@ -147,12 +147,11 @@ struct STsdbReader {
|
|||
static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter);
|
||||
static int buildDataBlockFromBufImpl(STableBlockScanInfo* pBlockScanInfo, TSDBKEY maxKey, int32_t capacity, STsdbReader* pReader);
|
||||
static TSDBROW* getValidRow(STbDataIter* pIter, bool* hasVal, STsdbReader* pReader);
|
||||
static int32_t doLoadRowsOfIdenticalTsInFileBlock(SFileDataBlockInfo* pFBlock, SBlock* pBlock, SBlockData* pBlockData,
|
||||
STableBlockScanInfo* pScanInfo, STsdbReader* pReader, SRowMerger* pMerger);
|
||||
static int32_t doLoadRowsOfIdenticalTsInFileBlock(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader, SRowMerger* pMerger);
|
||||
static int32_t doLoadRowsOfIdenticalTsInBuf(STbDataIter *pIter, bool* hasVal, int64_t ts, SRowMerger* pMerger, STsdbReader* pReader);
|
||||
static int32_t doAppendOneRow(SSDataBlock* pBlock, STsdbReader* pReader, STSRow* pTSRow);
|
||||
static void setComposedBlockFlag(STsdbReader* pReader, bool composed);
|
||||
static void checkUpdateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader);
|
||||
static void updateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader);
|
||||
|
||||
static void doMergeMultiRows(TSDBROW* pRow, uint64_t uid, STbDataIter* dIter, bool* hasVal, STSRow **pTSRow, STsdbReader* pReader);
|
||||
static void doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo *pBlockScanInfo, STsdbReader* pReader, STSRow** pTSRow);
|
||||
|
|
@ -1069,64 +1068,6 @@ _error:
|
|||
// return code;
|
||||
// }
|
||||
|
||||
// static int32_t loadFileDataBlock(STsdbReader* pTsdbReadHandle, SBlock* pBlock, STableBlockScanInfo* pCheckInfo,
|
||||
// bool* exists) {
|
||||
// SQueryFilePos* cur = &pTsdbReadHandle->cur;
|
||||
// int32_t code = TSDB_CODE_SUCCESS;
|
||||
// bool asc = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
|
||||
|
||||
// if (asc) {
|
||||
// // query ended in/started from current block
|
||||
// if (pTsdbReadHandle->window.ekey < pBlock->maxKey.ts || pCheckInfo->lastKey > pBlock->minKey.ts) {
|
||||
// if ((code = doLoadFileBlockData(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
|
||||
// *exists = false;
|
||||
// return code;
|
||||
// }
|
||||
|
||||
// SDataCols* pTSCol = pTsdbReadHandle->rhelper.pDCols[0];
|
||||
// assert(pTSCol->cols->type == TSDB_DATA_TYPE_TIMESTAMP && pTSCol->numOfRows == pBlock->numOfRows);
|
||||
|
||||
// if (pCheckInfo->lastKey > pBlock->minKey.ts) {
|
||||
// cur->pos =
|
||||
// binarySearchForKey(pTSCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey,
|
||||
// pTsdbReadHandle->order);
|
||||
// } else {
|
||||
// cur->pos = 0;
|
||||
// }
|
||||
|
||||
// assert(pCheckInfo->lastKey <= pBlock->maxKey.ts);
|
||||
// doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
|
||||
// } else { // the whole block is loaded in to buffer
|
||||
// cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
|
||||
// code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
|
||||
// }
|
||||
// } else { // desc order, query ended in current block
|
||||
// if (pTsdbReadHandle->window.ekey > pBlock->minKey.ts || pCheckInfo->lastKey < pBlock->maxKey.ts) {
|
||||
// if ((code = doLoadFileBlockData(pTsdbReadHandle, pBlock, pCheckInfo, cur->slot)) != TSDB_CODE_SUCCESS) {
|
||||
// *exists = false;
|
||||
// return code;
|
||||
// }
|
||||
|
||||
// SDataCols* pTsCol = pTsdbReadHandle->rhelper.pDCols[0];
|
||||
// if (pCheckInfo->lastKey < pBlock->maxKey.ts) {
|
||||
// cur->pos =
|
||||
// binarySearchForKey(pTsCol->cols[0].pData, pBlock->numOfRows, pCheckInfo->lastKey,
|
||||
// pTsdbReadHandle->order);
|
||||
// } else {
|
||||
// cur->pos = pBlock->numOfRows - 1;
|
||||
// }
|
||||
|
||||
// assert(pCheckInfo->lastKey >= pBlock->minKey.ts);
|
||||
// doMergeTwoLevelData(pTsdbReadHandle, pCheckInfo, pBlock);
|
||||
// } else {
|
||||
// cur->pos = asc ? 0 : (pBlock->numOfRows - 1);
|
||||
// code = handleDataMergeIfNeeded(pTsdbReadHandle, pBlock, pCheckInfo);
|
||||
// }
|
||||
// }
|
||||
|
||||
// *exists = pTsdbReadHandle->realNumOfRows > 0;
|
||||
// return code;
|
||||
// }
|
||||
|
||||
// static int doBinarySearchKey(char* pValue, int num, TSKEY key, int order) {
|
||||
// int firstPos, lastPos, midPos = -1;
|
||||
|
|
@ -1189,94 +1130,6 @@ _error:
|
|||
|
||||
// return midPos;
|
||||
// }
|
||||
|
||||
// static int32_t doCopyRowsFromFileBlock(STsdbReader* pTsdbReadHandle, int32_t capacity, int32_t numOfRows, int32_t
|
||||
// start,
|
||||
// int32_t end) {
|
||||
// SDataCols* pCols = pTsdbReadHandle->rhelper.pDCols[0];
|
||||
// TSKEY* tsArray = pCols->cols[0].pData;
|
||||
|
||||
// int32_t num = end - start + 1;
|
||||
// assert(num >= 0);
|
||||
|
||||
// if (num == 0) {
|
||||
// return numOfRows;
|
||||
// }
|
||||
|
||||
// bool ascScan = ASCENDING_TRAVERSE(pTsdbReadHandle->order);
|
||||
// int32_t trueStart = ascScan ? start : end;
|
||||
// int32_t trueEnd = ascScan ? end : start;
|
||||
// int32_t step = ascScan ? 1 : -1;
|
||||
|
||||
// int32_t requiredNumOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
|
||||
|
||||
// // data in buffer has greater timestamp, copy data in file block
|
||||
// int32_t i = 0, j = 0;
|
||||
// while (i < requiredNumOfCols && j < pCols->numOfCols) {
|
||||
// SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
|
||||
|
||||
// SDataCol* src = &pCols->cols[j];
|
||||
// if (src->colId < pColInfo->info.colId) {
|
||||
// j++;
|
||||
// continue;
|
||||
// }
|
||||
|
||||
// if (!isAllRowsNull(src) && pColInfo->info.colId == src->colId) {
|
||||
// if (!IS_VAR_DATA_TYPE(pColInfo->info.type)) { // todo opt performance
|
||||
// // memmove(pData, (char*)src->pData + bytes * start, bytes * num);
|
||||
// int32_t rowIndex = numOfRows;
|
||||
// for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex)
|
||||
// {
|
||||
// SCellVal sVal = {0};
|
||||
// if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
|
||||
// TASSERT(0);
|
||||
// }
|
||||
|
||||
// if (sVal.valType == TD_VTYPE_NORM) {
|
||||
// colDataAppend(pColInfo, rowIndex, sVal.val, false);
|
||||
// } else {
|
||||
// colDataAppendNULL(pColInfo, rowIndex);
|
||||
// }
|
||||
// }
|
||||
// } else { // handle the var-string
|
||||
// int32_t rowIndex = numOfRows;
|
||||
|
||||
// // todo refactor, only copy one-by-one
|
||||
// for (int32_t k = trueStart; ((ascScan && k <= trueEnd) || (!ascScan && k >= trueEnd)); k += step, ++rowIndex)
|
||||
// {
|
||||
// SCellVal sVal = {0};
|
||||
// if (tdGetColDataOfRow(&sVal, src, k, pCols->bitmapMode) < 0) {
|
||||
// TASSERT(0);
|
||||
// }
|
||||
|
||||
// if (sVal.valType == TD_VTYPE_NORM) {
|
||||
// colDataAppend(pColInfo, rowIndex, sVal.val, false);
|
||||
// } else {
|
||||
// colDataAppendNULL(pColInfo, rowIndex);
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
// j++;
|
||||
// i++;
|
||||
// } else { // pColInfo->info.colId < src->colId, it is a NULL data
|
||||
// colDataAppendNNULL(pColInfo, numOfRows, num);
|
||||
// i++;
|
||||
// }
|
||||
// }
|
||||
|
||||
// while (i < requiredNumOfCols) { // the remain columns are all null data
|
||||
// SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
|
||||
// colDataAppendNNULL(pColInfo, numOfRows, num);
|
||||
// i++;
|
||||
// }
|
||||
|
||||
// pTsdbReadHandle->cur.win.ekey = tsArray[trueEnd];
|
||||
// pTsdbReadHandle->cur.lastKey = tsArray[trueEnd] + step;
|
||||
|
||||
// return numOfRows + num;
|
||||
// }
|
||||
|
||||
// static int32_t mergeTwoRowFromMem(STsdbReader* pTsdbReadHandle, int32_t capacity, int32_t* curRow, STSRow* row1,
|
||||
// STSRow* row2, int32_t numOfCols, uint64_t uid, STSchema* pSchema1, STSchema*
|
||||
// pSchema2, bool update, TSKEY* lastRowKey) {
|
||||
|
|
@ -1472,42 +1325,6 @@ _error:
|
|||
// #endif
|
||||
// }
|
||||
|
||||
// static void getQualifiedRowsPos(STsdbReader* pTsdbReadHandle, int32_t startPos, int32_t endPos, int32_t numOfExisted,
|
||||
// int32_t* start, int32_t* end) {
|
||||
// *start = -1;
|
||||
|
||||
// if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
|
||||
// int32_t remain = endPos - startPos + 1;
|
||||
// if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
|
||||
// *end = (pTsdbReadHandle->outputCapacity - numOfExisted) + startPos - 1;
|
||||
// } else {
|
||||
// *end = endPos;
|
||||
// }
|
||||
|
||||
// *start = startPos;
|
||||
// } else {
|
||||
// int32_t remain = (startPos - endPos) + 1;
|
||||
// if (remain + numOfExisted > pTsdbReadHandle->outputCapacity) {
|
||||
// *end = startPos + 1 - (pTsdbReadHandle->outputCapacity - numOfExisted);
|
||||
// } else {
|
||||
// *end = endPos;
|
||||
// }
|
||||
|
||||
// *start = *end;
|
||||
// *end = startPos;
|
||||
// }
|
||||
// }
|
||||
|
||||
// static void updateInfoAfterMerge(STsdbReader* pTsdbReadHandle, STableBlockScanInfo* pCheckInfo, int32_t numOfRows,
|
||||
// int32_t endPos) {
|
||||
// SQueryFilePos* cur = &pTsdbReadHandle->cur;
|
||||
|
||||
// pCheckInfo->lastKey = cur->lastKey;
|
||||
// pTsdbReadHandle->realNumOfRows = numOfRows;
|
||||
// cur->rows = numOfRows;
|
||||
// cur->pos = endPos;
|
||||
// }
|
||||
|
||||
// static void doCheckGeneratedBlockRange(STsdbReader* pTsdbReadHandle) {
|
||||
// SQueryFilePos* cur = &pTsdbReadHandle->cur;
|
||||
|
||||
|
|
@ -2198,18 +2015,22 @@ static int32_t findFileBlockInfoIndex(SDataBlockIter* pBlockIter, SFileDataBlock
|
|||
return -1;
|
||||
}
|
||||
|
||||
static int32_t setFileBlockActiveInBlockIter(SDataBlockIter* pBlockIter, int32_t index) {
|
||||
static int32_t setFileBlockActiveInBlockIter(SDataBlockIter* pBlockIter, int32_t index, int32_t step) {
|
||||
if (index < 0 || index >= pBlockIter->numOfBlocks) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
SFileDataBlockInfo fblock = *(SFileDataBlockInfo*)taosArrayGet(pBlockIter->blockList, index);
|
||||
pBlockIter->index += step;
|
||||
|
||||
if (index != pBlockIter->index) {
|
||||
taosArrayRemove(pBlockIter->blockList, index);
|
||||
taosArrayInsert(pBlockIter->blockList, pBlockIter->index, &fblock);
|
||||
|
||||
SFileDataBlockInfo* pBlockInfo = taosArrayGet(pBlockIter->blockList, pBlockIter->index);
|
||||
ASSERT(pBlockInfo->uid == fblock.uid && pBlockInfo->tbBlockIdx == fblock.tbBlockIdx);
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
|
@ -2242,7 +2063,14 @@ static bool fileBlockShouldLoad(STsdbReader* pReader, SFileDataBlockInfo *pFBloc
|
|||
overlapWithNeighbor = overlapWithNeighborBlock(pBlock, pNeighbor, pReader->order);
|
||||
}
|
||||
|
||||
return (overlapWithNeighbor ||
|
||||
bool hasDup = false;
|
||||
if (pBlock->nSubBlock == 1) {
|
||||
hasDup = pBlock->hasDup;
|
||||
} else {
|
||||
hasDup = true;
|
||||
}
|
||||
|
||||
return (overlapWithNeighbor || hasDup ||
|
||||
dataBlockPartialRequired(&pReader->window, &pReader->verRange, pBlock) ||
|
||||
keyOverlapFileBlock(key, pBlock, &pReader->verRange) ||
|
||||
(pBlock->nRow > pReader->capacity));
|
||||
|
|
@ -2280,7 +2108,7 @@ static int32_t doMergeBufFileRows(STsdbReader* pReader, STableBlockScanInfo* pBl
|
|||
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
|
||||
tRowMergerInit(&merge, &fRow, pReader->pSchema);
|
||||
|
||||
doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
|
||||
doLoadRowsOfIdenticalTsInFileBlock(pBlockData, pBlockScanInfo, pReader, &merge);
|
||||
|
||||
if (k.ts == key) {
|
||||
tRowMerge(&merge, pRow);
|
||||
|
|
@ -2296,10 +2124,13 @@ static int32_t doMergeBufFileRows(STsdbReader* pReader, STableBlockScanInfo* pBl
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, SFileDataBlockInfo* pFBlock, SBlock* pBlock, STableBlockScanInfo* pBlockScanInfo) {
|
||||
SFileBlockDumpInfo *pDumpInfo = &pReader->status.fBlockDumpInfo;
|
||||
static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo) {
|
||||
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
|
||||
SBlockData* pBlockData = &pReader->status.fileBlockData;
|
||||
|
||||
SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter);
|
||||
SBlock* pBlock = taosArrayGet(pBlockScanInfo->pBlockList, pFBlock->tbBlockIdx);
|
||||
|
||||
SRowMerger merge = {0};
|
||||
STSRow* pTSRow = NULL;
|
||||
|
||||
|
|
@ -2316,7 +2147,7 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, SFileDataBlockIn
|
|||
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
|
||||
tRowMergerInit(&merge, &fRow, pReader->pSchema);
|
||||
|
||||
doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
|
||||
doLoadRowsOfIdenticalTsInFileBlock(pBlockData, pBlockScanInfo, pReader, &merge);
|
||||
|
||||
if (ik.ts == key) {
|
||||
tRowMerge(&merge, piRow);
|
||||
|
|
@ -2370,7 +2201,7 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, SFileDataBlockIn
|
|||
// imem & mem are all empty
|
||||
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
|
||||
tRowMergerInit(&merge, &fRow, pReader->pSchema);
|
||||
doLoadRowsOfIdenticalTsInFileBlock(pFBlock, pBlock, pBlockData, pBlockScanInfo, pReader, &merge);
|
||||
doLoadRowsOfIdenticalTsInFileBlock(pBlockData, pBlockScanInfo, pReader, &merge);
|
||||
tRowMergerGetRow(&merge, &pTSRow);
|
||||
doAppendOneRow(pReader->pResBlock, pReader, pTSRow);
|
||||
}
|
||||
|
|
@ -2378,16 +2209,17 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, SFileDataBlockIn
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t buildComposedDataBlock(STsdbReader* pReader, SFileDataBlockInfo* pFBlock, SBlock* pBlock,
|
||||
STableBlockScanInfo* pBlockScanInfo) {
|
||||
static int32_t buildComposedDataBlock(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo) {
|
||||
SSDataBlock* pResBlock = pReader->pResBlock;
|
||||
|
||||
while(1) {
|
||||
buildComposedDataBlockImpl(pReader, pFBlock, pBlock, pBlockScanInfo);
|
||||
buildComposedDataBlockImpl(pReader, pBlockScanInfo);
|
||||
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
|
||||
|
||||
SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
|
||||
if (pBlockInfo->tbBlockIdx == pFBlock->tbBlockIdx) { // still in the same file block now
|
||||
SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter);
|
||||
SBlock* pBlock = taosArrayGet(pBlockScanInfo->pBlockList, pFBlock->tbBlockIdx);
|
||||
|
||||
// currently loaded file data block is consumed
|
||||
if (pDumpInfo->rowIndex >= pBlock->nRow || pDumpInfo->rowIndex < 0) {
|
||||
setBlockAllDumped(pDumpInfo, pBlock, pReader->order);
|
||||
break;
|
||||
|
|
@ -2396,16 +2228,16 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader, SFileDataBlockInfo*
|
|||
if (pResBlock->info.rows >= pReader->capacity) {
|
||||
break;
|
||||
}
|
||||
} else { // todo traverse to next file due to time window overlap
|
||||
if (pResBlock->info.rows >= pReader->capacity) {
|
||||
ASSERT(0);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pResBlock->info.uid = pBlockScanInfo->uid;
|
||||
blockDataUpdateTsWindow(pResBlock, 0);
|
||||
|
||||
setComposedBlockFlag(pReader, true);
|
||||
|
||||
tsdbDebug("%p uid:%" PRIu64 ", composed data block created, brange:%" PRIu64 "-%" PRIu64 " rows:%d, %s",
|
||||
pReader, pBlockScanInfo->uid, pResBlock->info.window.skey, pResBlock->info.window.ekey, pResBlock->info.rows,
|
||||
pReader->idStr);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
|
@ -2549,7 +2381,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
|
|||
}
|
||||
|
||||
// build composed data block
|
||||
code = buildComposedDataBlock(pReader, pFBlock, pBlock, pScanInfo);
|
||||
code = buildComposedDataBlock(pReader, pScanInfo);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
return code;
|
||||
}
|
||||
|
|
@ -2653,7 +2485,6 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
|
|||
} else {
|
||||
SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter);
|
||||
STableBlockScanInfo* pScanInfo = taosHashGet(pReader->status.pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
|
||||
SBlock* pBlock = taosArrayGet(pScanInfo->pBlockList, pFBlock->tbBlockIdx);
|
||||
|
||||
// current block are exhausted, try the next file block
|
||||
if (pReader->status.fBlockDumpInfo.allDumped) {
|
||||
|
|
@ -2673,7 +2504,7 @@ static int32_t buildBlockFromFiles(STsdbReader* pReader) {
|
|||
return code;
|
||||
}
|
||||
} else {
|
||||
code = buildComposedDataBlock(pReader, pFBlock, pBlock, pScanInfo);
|
||||
code = buildComposedDataBlock(pReader, pScanInfo);
|
||||
return code;
|
||||
}
|
||||
}
|
||||
|
|
@ -2815,10 +2646,10 @@ static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanIn
|
|||
|
||||
//1. find the next neighbor block in the scan block list
|
||||
SFileDataBlockInfo fb = {.uid = pFBlock->uid, .tbBlockIdx = nextIndex};
|
||||
int32_t neighborIndex = findFileBlockInfoIndex(&pStatus->blockIter, &fb);
|
||||
int32_t neighborIndex = findFileBlockInfoIndex(pBlockIter, &fb);
|
||||
|
||||
//2. remove it from the scan block list
|
||||
setFileBlockActiveInBlockIter(&pStatus->blockIter, neighborIndex);
|
||||
setFileBlockActiveInBlockIter(pBlockIter, neighborIndex, step);
|
||||
|
||||
//3. load the neighbor block, and set it to be the currently accessed file data block
|
||||
int32_t code = doLoadFileBlockData(pReader, pBlockIter, pScanInfo, &pStatus->fileBlockData);
|
||||
|
|
@ -2840,17 +2671,16 @@ static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanIn
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
int32_t doLoadRowsOfIdenticalTsInFileBlock(SFileDataBlockInfo* pFBlock, SBlock* pBlock, SBlockData* pBlockData,
|
||||
STableBlockScanInfo* pScanInfo, STsdbReader* pReader, SRowMerger* pMerger) {
|
||||
int32_t doLoadRowsOfIdenticalTsInFileBlock(SBlockData* pBlockData, STableBlockScanInfo* pScanInfo, STsdbReader* pReader, SRowMerger* pMerger) {
|
||||
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
|
||||
|
||||
bool asc = ASCENDING_TRAVERSE(pReader->order);
|
||||
int32_t step = asc ? 1 : -1;
|
||||
int64_t key = pBlockData->aTSKEY[pDumpInfo->rowIndex];
|
||||
int32_t step = asc ? 1 : -1;
|
||||
|
||||
if (asc) {
|
||||
pDumpInfo->rowIndex += step;
|
||||
if (pDumpInfo->rowIndex < pBlockData->nRow - 1) {
|
||||
if (pDumpInfo->rowIndex <= pBlockData->nRow - 1) {
|
||||
pDumpInfo->rowIndex = doMergeRowsInFileBlockImpl(pBlockData, pDumpInfo->rowIndex, key, pMerger, &pReader->verRange, step);
|
||||
}
|
||||
|
||||
|
|
@ -2858,7 +2688,10 @@ int32_t doLoadRowsOfIdenticalTsInFileBlock(SFileDataBlockInfo* pFBlock, SBlock*
|
|||
if (pDumpInfo->rowIndex >= pBlockData->nRow && asc) {
|
||||
while (1) {
|
||||
CHECK_FILEBLOCK_STATE st;
|
||||
checkForNeighborFileBlock(pReader, pScanInfo, pBlock, pFBlock, pMerger, key, &st);
|
||||
|
||||
SFileDataBlockInfo* pFileBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
|
||||
SBlock* pCurrentBlock = taosArrayGet(pScanInfo->pBlockList, pFileBlockInfo->tbBlockIdx);
|
||||
checkForNeighborFileBlock(pReader, pScanInfo, pCurrentBlock, pFileBlockInfo, pMerger, key, &st);
|
||||
if (st == CHECK_FILEBLOCK_QUIT) {
|
||||
break;
|
||||
}
|
||||
|
|
@ -2876,7 +2709,7 @@ int32_t doLoadRowsOfIdenticalTsInFileBlock(SFileDataBlockInfo* pFBlock, SBlock*
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
void checkUpdateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader) {
|
||||
void updateSchema(TSDBROW* pRow, uint64_t uid, STsdbReader* pReader) {
|
||||
int32_t sversion = TSDBROW_SVERSION(pRow);
|
||||
|
||||
if (pReader->pSchema == NULL) {
|
||||
|
|
@ -2891,7 +2724,7 @@ void doMergeMultiRows(TSDBROW* pRow, uint64_t uid, STbDataIter* dIter, bool* has
|
|||
SRowMerger merge = {0};
|
||||
|
||||
TSDBKEY k = TSDBROW_KEY(pRow);
|
||||
checkUpdateSchema(pRow, uid, pReader);
|
||||
updateSchema(pRow, uid, pReader);
|
||||
|
||||
tRowMergerInit(&merge, pRow, pReader->pSchema);
|
||||
doLoadRowsOfIdenticalTsInBuf(dIter, hasVal, k.ts, &merge, pReader);
|
||||
|
|
@ -2905,7 +2738,7 @@ void doMergeMemIMemRows(TSDBROW* pRow, TSDBROW* piRow, STableBlockScanInfo *pBlo
|
|||
TSDBKEY ik = TSDBROW_KEY(piRow);
|
||||
ASSERT(k.ts == ik.ts);
|
||||
|
||||
checkUpdateSchema(piRow, pBlockScanInfo->uid, pReader);
|
||||
updateSchema(piRow, pBlockScanInfo->uid, pReader);
|
||||
|
||||
tRowMergerInit(&merge, piRow, pReader->pSchema);
|
||||
doLoadRowsOfIdenticalTsInBuf(pBlockScanInfo->iiter, &pBlockScanInfo->imemHasVal, ik.ts, &merge, pReader);
|
||||
|
|
|
|||
|
|
@ -1139,9 +1139,55 @@ _err:
|
|||
return code;
|
||||
}
|
||||
|
||||
int32_t tsdbReadBlockSMA(SDataFReader *pReader, SBlockSMA *pBlkSMA) {
|
||||
int32_t tsdbReadBlockSma(SDataFReader *pReader, SBlock *pBlock, SArray *aColumnDataAgg, uint8_t **ppBuf) {
|
||||
int32_t code = 0;
|
||||
// TODO
|
||||
TdFilePtr pFD = pReader->pSmaFD;
|
||||
int64_t offset = pBlock->aSubBlock[0].offset;
|
||||
int64_t size = pBlock->aSubBlock[0].nSma * sizeof(SColumnDataAgg) + sizeof(TSCKSUM);
|
||||
uint8_t *pBuf = NULL;
|
||||
int64_t n;
|
||||
|
||||
ASSERT(tBlockHasSma(pBlock));
|
||||
|
||||
if (!ppBuf) ppBuf = &pBuf;
|
||||
code = tsdbRealloc(ppBuf, size);
|
||||
if (code) goto _err;
|
||||
|
||||
// lseek
|
||||
n = taosLSeekFile(pFD, offset, SEEK_SET);
|
||||
if (n < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
goto _err;
|
||||
}
|
||||
|
||||
// read
|
||||
n = taosReadFile(pFD, *ppBuf, size);
|
||||
if (n < 0) {
|
||||
code = TAOS_SYSTEM_ERROR(errno);
|
||||
goto _err;
|
||||
}
|
||||
|
||||
// check
|
||||
if (!taosCheckChecksumWhole(NULL, size)) {
|
||||
code = TSDB_CODE_FILE_CORRUPTED;
|
||||
goto _err;
|
||||
}
|
||||
|
||||
// decode
|
||||
taosArrayClear(aColumnDataAgg);
|
||||
for (int32_t iSma = 0; iSma < pBlock->aSubBlock[0].nSma; iSma++) {
|
||||
if (taosArrayPush(aColumnDataAgg, &((SColumnDataAgg *)(*ppBuf))[iSma]) == NULL) {
|
||||
code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
goto _err;
|
||||
}
|
||||
}
|
||||
|
||||
tsdbFree(pBuf);
|
||||
return code;
|
||||
|
||||
_err:
|
||||
tsdbError("vgId:%d read block sma failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code));
|
||||
tsdbFree(pBuf);
|
||||
return code;
|
||||
}
|
||||
|
||||
|
|
@ -1699,24 +1745,7 @@ _err:
|
|||
return code;
|
||||
}
|
||||
|
||||
static void tsdbCalcColDataSMA(SColData *pColData, SColumnDataAgg *pColAgg) {
|
||||
SColVal colVal;
|
||||
SColVal *pColVal = &colVal;
|
||||
|
||||
*pColAgg = (SColumnDataAgg){.colId = pColData->cid};
|
||||
for (int32_t iVal = 0; iVal < pColData->nVal; iVal++) {
|
||||
tColDataGetValue(pColData, iVal, pColVal);
|
||||
|
||||
if (pColVal->isNone || pColVal->isNull) {
|
||||
pColAgg->numOfNull++;
|
||||
} else {
|
||||
// TODO:
|
||||
ASSERT(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static int32_t tsdbWriteBlockSMA(TdFilePtr pFD, SBlockData *pBlockData, SSubBlock *pSubBlock, uint8_t **ppBuf) {
|
||||
static int32_t tsdbWriteBlockSma(TdFilePtr pFD, SBlockData *pBlockData, SSubBlock *pSubBlock, uint8_t **ppBuf) {
|
||||
int32_t code = 0;
|
||||
int64_t n;
|
||||
SColData *pColData;
|
||||
|
|
@ -1843,7 +1872,7 @@ int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, uint8_
|
|||
|
||||
if (pBlock->nSubBlock > 1 || pBlock->last || pBlock->hasDup) goto _exit;
|
||||
|
||||
code = tsdbWriteBlockSMA(pWriter->pSmaFD, pBlockData, pSubBlock, ppBuf1);
|
||||
code = tsdbWriteBlockSma(pWriter->pSmaFD, pBlockData, pSubBlock, ppBuf1);
|
||||
if (code) goto _err;
|
||||
|
||||
if (pSubBlock->nSma > 0) {
|
||||
|
|
|
|||
|
|
@ -353,6 +353,14 @@ int32_t tBlockCmprFn(const void *p1, const void *p2) {
|
|||
return 0;
|
||||
}
|
||||
|
||||
bool tBlockHasSma(SBlock *pBlock) {
|
||||
if (pBlock->nSubBlock > 1) return false;
|
||||
if (pBlock->last) return false;
|
||||
if (pBlock->hasDup) return false;
|
||||
|
||||
return pBlock->aSubBlock[0].nSma > 0;
|
||||
}
|
||||
|
||||
// SBlockCol ======================================================
|
||||
int32_t tPutBlockCol(uint8_t *p, void *ph) {
|
||||
int32_t n = 0;
|
||||
|
|
@ -769,29 +777,6 @@ int32_t tsdbBuildDeleteSkyline(SArray *aDelData, int32_t sidx, int32_t eidx, SAr
|
|||
return code;
|
||||
}
|
||||
|
||||
// KEYINFO ======================================================
|
||||
int32_t tPutKEYINFO(uint8_t *p, KEYINFO *pKeyInfo) {
|
||||
int32_t n = 0;
|
||||
|
||||
n += tPutTSDBKEY(p ? p + n : p, &pKeyInfo->minKey);
|
||||
n += tPutTSDBKEY(p ? p + n : p, &pKeyInfo->maxKey);
|
||||
n += tPutI64v(p ? p + n : p, pKeyInfo->minVerion);
|
||||
n += tPutI64v(p ? p + n : p, pKeyInfo->maxVersion);
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
int32_t tGetKEYINFO(uint8_t *p, KEYINFO *pKeyInfo) {
|
||||
int32_t n = 0;
|
||||
|
||||
n += tGetTSDBKEY(p + n, &pKeyInfo->minKey);
|
||||
n += tGetTSDBKEY(p + n, &pKeyInfo->maxKey);
|
||||
n += tGetI64v(p + n, &pKeyInfo->minVerion);
|
||||
n += tGetI64v(p + n, &pKeyInfo->maxVersion);
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
// SColData ========================================
|
||||
void tColDataReset(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn) {
|
||||
pColData->cid = cid;
|
||||
|
|
@ -1192,3 +1177,63 @@ void tBlockDataGetColData(SBlockData *pBlockData, int16_t cid, SColData **ppColD
|
|||
*ppColData = *(SColData **)p;
|
||||
}
|
||||
}
|
||||
|
||||
// ALGORITHM ==============================
|
||||
void tsdbCalcColDataSMA(SColData *pColData, SColumnDataAgg *pColAgg) {
|
||||
SColVal colVal;
|
||||
SColVal *pColVal = &colVal;
|
||||
|
||||
*pColAgg = (SColumnDataAgg){.colId = pColData->cid};
|
||||
for (int32_t iVal = 0; iVal < pColData->nVal; iVal++) {
|
||||
tColDataGetValue(pColData, iVal, pColVal);
|
||||
|
||||
if (pColVal->isNone || pColVal->isNull) {
|
||||
pColAgg->numOfNull++;
|
||||
} else {
|
||||
switch (pColData->type) {
|
||||
case TSDB_DATA_TYPE_NULL:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_BOOL:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_TINYINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_SMALLINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_INT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_BIGINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_FLOAT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_DOUBLE:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_VARCHAR:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_TIMESTAMP:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_NCHAR:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_UTINYINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_USMALLINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_UINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_UBIGINT:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_JSON:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_VARBINARY:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_DECIMAL:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_BLOB:
|
||||
break;
|
||||
case TSDB_DATA_TYPE_MEDIUMBLOB:
|
||||
break;
|
||||
default:
|
||||
ASSERT(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue