Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 e17c65fd 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch 'enh/load_by_column' of https://github.com/taosdata/TDengine into... 

Merge branch 'enh/load_by_column' of https://github.com/taosdata/TDengine into feat/stream_compression
上级 3be4b1d4 144a0c57
隐藏空白更改
内联 并排
Showing with 145 addition and 59 deletion
source/dnode/vnode/src/inc/tsdb.h
浏览文件 @ e17c65fd
...@@ -159,7 +159,7 @@ int32_t tCmprBlockL(void const *lhs, void const *rhs); ...@@ -159,7 +159,7 @@ int32_t tCmprBlockL(void const *lhs, void const *rhs);
int32_t tBlockDataCreate(SBlockData *pBlockData); int32_t tBlockDataCreate(SBlockData *pBlockData);
void tBlockDataDestroy(SBlockData *pBlockData, int8_t deepClear); void tBlockDataDestroy(SBlockData *pBlockData, int8_t deepClear);
int32_t tBlockDataInit(SBlockData *pBlockData, int64_t suid, int64_t uid, STSchema *pTSchema); int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema, int16_t *aCid, int32_t nCid);
int32_t tBlockDataInitEx(SBlockData *pBlockData, SBlockData *pBlockDataFrom); int32_t tBlockDataInitEx(SBlockData *pBlockData, SBlockData *pBlockDataFrom);
void tBlockDataReset(SBlockData *pBlockData); void tBlockDataReset(SBlockData *pBlockData);
int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid); int32_t tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema, int64_t uid);
...@@ -281,6 +281,7 @@ int32_t tsdbReadSttBlk(SDataFReader *pReader, int32_t iStt, SArray *aSttBlk); ...@@ -281,6 +281,7 @@ int32_t tsdbReadSttBlk(SDataFReader *pReader, int32_t iStt, SArray *aSttBlk);
int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pBlock, SArray *aColumnDataAgg); int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pBlock, SArray *aColumnDataAgg);
int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pBlock, SBlockData *pBlockData); int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pBlock, SBlockData *pBlockData);
int32_t tsdbReadSttBlock(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData); int32_t tsdbReadSttBlock(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData);
int32_t tsdbReadSttBlockEx(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData);
// SDelFWriter // SDelFWriter
int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb); int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb);
int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync); int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync);
...@@ -651,6 +652,9 @@ typedef struct SSttBlockLoadInfo { ...@@ -651,6 +652,9 @@ typedef struct SSttBlockLoadInfo {
int32_t currentLoadBlockIndex; int32_t currentLoadBlockIndex;
int32_t loadBlocks; int32_t loadBlocks;
double elapsedTime; double elapsedTime;
STSchema *pSchema;
int16_t *colIds;
int32_t numOfCols;
} SSttBlockLoadInfo; } SSttBlockLoadInfo;
typedef struct SMergeTree { typedef struct SMergeTree {
...@@ -701,13 +705,14 @@ struct SDiskDataBuilder { ...@@ -701,13 +705,14 @@ struct SDiskDataBuilder {
}; };
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid, int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
STimeWindow *pTimeWindow, SVersionRange *pVerRange, void *pLoadInfo, const char *idStr); STimeWindow *pTimeWindow, SVersionRange *pVerRange, void *pBlockLoadInfo, STSchema *pSchema,
int16_t *pCols, int32_t numOfCols, const char *idStr);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter); void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree); bool tMergeTreeNext(SMergeTree *pMTree);
TSDBROW tMergeTreeGetRow(SMergeTree *pMTree); TSDBROW tMergeTreeGetRow(SMergeTree *pMTree);
void tMergeTreeClose(SMergeTree *pMTree); void tMergeTreeClose(SMergeTree *pMTree);
SSttBlockLoadInfo *tCreateLastBlockLoadInfo(); SSttBlockLoadInfo *tCreateLastBlockLoadInfo(STSchema *pSchema, int16_t *colList, int32_t numOfCols);
void resetLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo); void resetLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
void getLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo, int64_t *blocks, double *el); void getLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo, int64_t *blocks, double *el);
void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo); void *destroyLastBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo);
......
source/dnode/vnode/src/tsdb/tsdbCache.c
浏览文件 @ e17c65fd
...@@ -457,7 +457,7 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) { ...@@ -457,7 +457,7 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
tMergeTreeOpen(&state->mergeTree, 1, state->pDataFReader, state->suid, state->uid, tMergeTreeOpen(&state->mergeTree, 1, state->pDataFReader, state->suid, state->uid,
&(STimeWindow){.skey = TSKEY_MIN, .ekey = TSKEY_MAX}, &(STimeWindow){.skey = TSKEY_MIN, .ekey = TSKEY_MAX},
&(SVersionRange){.minVer = 0, .maxVer = UINT64_MAX}, NULL, NULL); &(SVersionRange){.minVer = 0, .maxVer = UINT64_MAX}, NULL, NULL, NULL, 0, NULL);
bool hasVal = tMergeTreeNext(&state->mergeTree); bool hasVal = tMergeTreeNext(&state->mergeTree);
if (!hasVal) { if (!hasVal) {
state->state = SFSLASTNEXTROW_FILESET; state->state = SFSLASTNEXTROW_FILESET;
...@@ -612,7 +612,8 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) { ...@@ -612,7 +612,8 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
tMapDataGetItemByIdx(&state->blockMap, state->iBlock, &block, tGetDataBlk); tMapDataGetItemByIdx(&state->blockMap, state->iBlock, &block, tGetDataBlk);
/* code = tsdbReadBlockData(state->pDataFReader, &state->blockIdx, &block, &state->blockData, NULL, NULL); */ /* code = tsdbReadBlockData(state->pDataFReader, &state->blockIdx, &block, &state->blockData, NULL, NULL); */
tBlockDataReset(state->pBlockData); tBlockDataReset(state->pBlockData);
code = tBlockDataInit(state->pBlockData, state->suid, state->uid, state->pTSchema); TABLEID tid = {.suid = state->suid, .uid = state->uid};
code = tBlockDataInit(state->pBlockData, &tid, state->pTSchema, NULL, 0);
if (code) goto _err; if (code) goto _err;
code = tsdbReadDataBlock(state->pDataFReader, &block, state->pBlockData); code = tsdbReadDataBlock(state->pDataFReader, &block, state->pBlockData);
......
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ e17c65fd
...@@ -439,8 +439,8 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) { ...@@ -439,8 +439,8 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) {
pIter->iSttBlk = 0; pIter->iSttBlk = 0;
SSttBlk *pSttBlk = (SSttBlk *)taosArrayGet(pIter->aSttBlk, 0); SSttBlk *pSttBlk = (SSttBlk *)taosArrayGet(pIter->aSttBlk, 0);
code = tsdbReadSttBlock(pCommitter->dReader.pReader, iStt, pSttBlk, &pIter->bData); code = tsdbReadSttBlockEx(pCommitter->dReader.pReader, iStt, pSttBlk, &pIter->bData);
TSDB_CHECK_CODE(code, lino, _exit); if (code) goto _err;
pIter->iRow = 0; pIter->iRow = 0;
pIter->r.suid = pIter->bData.suid; pIter->r.suid = pIter->bData.suid;
...@@ -1101,8 +1101,8 @@ static int32_t tsdbNextCommitRow(SCommitter *pCommitter) { ...@@ -1101,8 +1101,8 @@ static int32_t tsdbNextCommitRow(SCommitter *pCommitter) {
if (pIter->iSttBlk < taosArrayGetSize(pIter->aSttBlk)) { if (pIter->iSttBlk < taosArrayGetSize(pIter->aSttBlk)) {
SSttBlk *pSttBlk = (SSttBlk *)taosArrayGet(pIter->aSttBlk, pIter->iSttBlk); SSttBlk *pSttBlk = (SSttBlk *)taosArrayGet(pIter->aSttBlk, pIter->iSttBlk);
code = tsdbReadSttBlock(pCommitter->dReader.pReader, pIter->iStt, pSttBlk, &pIter->bData); code = tsdbReadSttBlockEx(pCommitter->dReader.pReader, pIter->iStt, pSttBlk, &pIter->bData);
TSDB_CHECK_CODE(code, lino, _exit); if (code) goto _exit;
pIter->iRow = 0; pIter->iRow = 0;
pIter->r.suid = pIter->bData.suid; pIter->r.suid = pIter->bData.suid;
...@@ -1369,8 +1369,9 @@ static int32_t tsdbInitSttBlockBuilderIfNeed(SCommitter *pCommitter, TABLEID id) ...@@ -1369,8 +1369,9 @@ static int32_t tsdbInitSttBlockBuilderIfNeed(SCommitter *pCommitter, TABLEID id)
if (!pBuilder->suid && !pBuilder->uid) { if (!pBuilder->suid && !pBuilder->uid) {
ASSERT(pCommitter->skmTable.suid == id.suid); ASSERT(pCommitter->skmTable.suid == id.suid);
ASSERT(pCommitter->skmTable.uid == id.uid); ASSERT(pCommitter->skmTable.uid == id.uid);
code = tDiskDataBuilderInit(pBuilder, pCommitter->skmTable.pTSchema, &id, pCommitter->cmprAlg, 0); TABLEID tid = {.suid = id.suid, .uid = id.suid ? 0 : id.uid};
TSDB_CHECK_CODE(code, lino, _exit); code = tBlockDataInit(pBDatal, &tid, pCommitter->skmTable.pTSchema, NULL, 0);
if (code) goto _exit;
} }
_exit: _exit:
...@@ -1518,11 +1519,11 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) { ...@@ -1518,11 +1519,11 @@ static int32_t tsdbCommitFileDataImpl(SCommitter *pCommitter) {
// impl // impl
code = tsdbUpdateTableSchema(pCommitter->pTsdb->pVnode->pMeta, id.suid, id.uid, &pCommitter->skmTable); code = tsdbUpdateTableSchema(pCommitter->pTsdb->pVnode->pMeta, id.suid, id.uid, &pCommitter->skmTable);
TSDB_CHECK_CODE(code, lino, _exit); if (code) goto _err;
code = tBlockDataInit(&pCommitter->dReader.bData, id.suid, id.uid, pCommitter->skmTable.pTSchema); code = tBlockDataInit(&pCommitter->dReader.bData, &id, pCommitter->skmTable.pTSchema, NULL, 0);
TSDB_CHECK_CODE(code, lino, _exit); if (code) goto _err;
code = tBlockDataInit(&pCommitter->dWriter.bData, id.suid, id.uid, pCommitter->skmTable.pTSchema); code = tBlockDataInit(&pCommitter->dWriter.bData, &id, pCommitter->skmTable.pTSchema, NULL, 0);
TSDB_CHECK_CODE(code, lino, _exit); if (code) goto _err;
/* merge with data in .data file */ /* merge with data in .data file */
code = tsdbMergeTableData(pCommitter, id); code = tsdbMergeTableData(pCommitter, id);
......
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
浏览文件 @ e17c65fd
...@@ -28,11 +28,10 @@ struct SLDataIter { ...@@ -28,11 +28,10 @@ struct SLDataIter {
uint64_t uid; uint64_t uid;
STimeWindow timeWindow; STimeWindow timeWindow;
SVersionRange verRange; SVersionRange verRange;
SSttBlockLoadInfo* pBlockLoadInfo; SSttBlockLoadInfo* pBlockLoadInfo;
}; };
SSttBlockLoadInfo* tCreateLastBlockLoadInfo() { SSttBlockLoadInfo* tCreateLastBlockLoadInfo(STSchema* pSchema, int16_t* colList, int32_t numOfCols) {
SSttBlockLoadInfo* pLoadInfo = taosMemoryCalloc(TSDB_DEFAULT_STT_FILE, sizeof(SSttBlockLoadInfo)); SSttBlockLoadInfo* pLoadInfo = taosMemoryCalloc(TSDB_DEFAULT_STT_FILE, sizeof(SSttBlockLoadInfo));
if (pLoadInfo == NULL) { if (pLoadInfo == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
...@@ -55,6 +54,9 @@ SSttBlockLoadInfo* tCreateLastBlockLoadInfo() { ...@@ -55,6 +54,9 @@ SSttBlockLoadInfo* tCreateLastBlockLoadInfo() {
} }
pLoadInfo[i].aSttBlk = taosArrayInit(4, sizeof(SSttBlk)); pLoadInfo[i].aSttBlk = taosArrayInit(4, sizeof(SSttBlk));
pLoadInfo[i].pSchema = pSchema;
pLoadInfo[i].colIds = colList;
pLoadInfo[i].numOfCols = numOfCols;
} }
return pLoadInfo; return pLoadInfo;
...@@ -111,7 +113,19 @@ static SBlockData* loadLastBlock(SLDataIter *pIter, const char* idStr) { ...@@ -111,7 +113,19 @@ static SBlockData* loadLastBlock(SLDataIter *pIter, const char* idStr) {
pInfo->currentLoadBlockIndex ^= 1; pInfo->currentLoadBlockIndex ^= 1;
if (pIter->pSttBlk != NULL) { // current block not loaded yet if (pIter->pSttBlk != NULL) { // current block not loaded yet
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
code = tsdbReadSttBlock(pIter->pReader, pIter->iStt, pIter->pSttBlk, &pInfo->blockData[pInfo->currentLoadBlockIndex]);
SBlockData* pBlock = &pInfo->blockData[pInfo->currentLoadBlockIndex];
TABLEID id = {0};
if (pIter->pSttBlk->suid != 0) {
id.suid = pIter->pSttBlk->suid;
} else {
id.uid = pIter->uid;
}
tBlockDataInit(pBlock, &id, pInfo->pSchema, pInfo->colIds, pInfo->numOfCols);
code = tsdbReadSttBlock(pIter->pReader, pIter->iStt, pIter->pSttBlk, pBlock);
double el = (taosGetTimestampUs() - st)/ 1000.0; double el = (taosGetTimestampUs() - st)/ 1000.0;
pInfo->elapsedTime += el; pInfo->elapsedTime += el;
pInfo->loadBlocks += 1; pInfo->loadBlocks += 1;
...@@ -460,7 +474,8 @@ static FORCE_INLINE int32_t tLDataIterCmprFn(const void *p1, const void *p2) { ...@@ -460,7 +474,8 @@ static FORCE_INLINE int32_t tLDataIterCmprFn(const void *p1, const void *p2) {
} }
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid, int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
STimeWindow *pTimeWindow, SVersionRange *pVerRange, void* pBlockLoadInfo, const char* idStr) { STimeWindow *pTimeWindow, SVersionRange *pVerRange, void* pBlockLoadInfo, STSchema* pSchema,
int16_t* pCols, int32_t numOfCols, const char* idStr) {
pMTree->backward = backward; pMTree->backward = backward;
pMTree->pIter = NULL; pMTree->pIter = NULL;
pMTree->pIterList = taosArrayInit(4, POINTER_BYTES); pMTree->pIterList = taosArrayInit(4, POINTER_BYTES);
...@@ -475,9 +490,10 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead ...@@ -475,9 +490,10 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead
SSttBlockLoadInfo* pLoadInfo = NULL; SSttBlockLoadInfo* pLoadInfo = NULL;
if (pBlockLoadInfo == NULL) { if (pBlockLoadInfo == NULL) {
ASSERT(0);
if (pMTree->pLoadInfo == NULL) { if (pMTree->pLoadInfo == NULL) {
pMTree->destroyLoadInfo = true; pMTree->destroyLoadInfo = true;
pMTree->pLoadInfo = tCreateLastBlockLoadInfo(); pMTree->pLoadInfo = tCreateLastBlockLoadInfo(pSchema, pCols, numOfCols);
} }
pLoadInfo = pMTree->pLoadInfo; pLoadInfo = pMTree->pLoadInfo;
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ e17c65fd
...@@ -79,6 +79,7 @@ typedef struct SBlockLoadSuppInfo { ...@@ -79,6 +79,7 @@ typedef struct SBlockLoadSuppInfo {
SColumnDataAgg tsColAgg; SColumnDataAgg tsColAgg;
SColumnDataAgg** plist; SColumnDataAgg** plist;
int16_t* colIds; // column ids for loading file block data int16_t* colIds; // column ids for loading file block data
int32_t numOfCols;
char** buildBuf; // build string tmp buffer, todo remove it later after all string format being updated. char** buildBuf; // build string tmp buffer, todo remove it later after all string format being updated.
} SBlockLoadSuppInfo; } SBlockLoadSuppInfo;
...@@ -203,6 +204,7 @@ static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) { ...@@ -203,6 +204,7 @@ static int32_t setColumnIdSlotList(STsdbReader* pReader, SSDataBlock* pBlock) {
size_t numOfCols = blockDataGetNumOfCols(pBlock); size_t numOfCols = blockDataGetNumOfCols(pBlock);
pSupInfo->numOfCols = numOfCols;
pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t)); pSupInfo->colIds = taosMemoryMalloc(numOfCols * sizeof(int16_t));
pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES); pSupInfo->buildBuf = taosMemoryCalloc(numOfCols, POINTER_BYTES);
if (pSupInfo->buildBuf == NULL || pSupInfo->colIds == NULL) { if (pSupInfo->buildBuf == NULL || pSupInfo->colIds == NULL) {
...@@ -352,7 +354,8 @@ static int32_t initFilesetIterator(SFilesetIter* pIter, SArray* aDFileSet, STsdb ...@@ -352,7 +354,8 @@ static int32_t initFilesetIterator(SFilesetIter* pIter, SArray* aDFileSet, STsdb
tMergeTreeClose(&pLReader->mergeTree); tMergeTreeClose(&pLReader->mergeTree);
if (pLReader->pInfo == NULL) { if (pLReader->pInfo == NULL) {
pLReader->pInfo = tCreateLastBlockLoadInfo(); // here we ignore the first column, which is always be the primary timestamp column
pLReader->pInfo = tCreateLastBlockLoadInfo(pReader->pSchema, &pReader->suppInfo.colIds[1], pReader->suppInfo.numOfCols - 1);
if (pLReader->pInfo == NULL) { if (pLReader->pInfo == NULL) {
tsdbDebug("init fileset iterator failed, code:%s %s", tstrerror(terrno), pReader->idStr); tsdbDebug("init fileset iterator failed, code:%s %s", tstrerror(terrno), pReader->idStr);
return terrno; return terrno;
...@@ -1741,7 +1744,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* ...@@ -1741,7 +1744,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
tsLast = getCurrentKeyInLastBlock(pLastBlockReader); tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
} }
int64_t key = hasDataInFileBlock(pBlockData, pDumpInfo)? pBlockData->aTSKEY[pDumpInfo->rowIndex]:INT64_MIN; int64_t key = hasDataInFileBlock(pBlockData, pDumpInfo) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
TSDBKEY k = TSDBROW_KEY(pRow); TSDBKEY k = TSDBROW_KEY(pRow);
TSDBKEY ik = TSDBROW_KEY(piRow); TSDBKEY ik = TSDBROW_KEY(piRow);
...@@ -1995,7 +1998,8 @@ static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScan ...@@ -1995,7 +1998,8 @@ static bool initLastBlockReader(SLastBlockReader* pLBlockReader, STableBlockScan
int32_t code = int32_t code =
tMergeTreeOpen(&pLBlockReader->mergeTree, (pLBlockReader->order == TSDB_ORDER_DESC), pReader->pFileReader, tMergeTreeOpen(&pLBlockReader->mergeTree, (pLBlockReader->order == TSDB_ORDER_DESC), pReader->pFileReader,
pReader->suid, pScanInfo->uid, &w, &pLBlockReader->verRange, pLBlockReader->pInfo, pReader->idStr); pReader->suid, pScanInfo->uid, &w, &pLBlockReader->verRange, pLBlockReader->pInfo,
pReader->pSchema, pReader->suppInfo.colIds, pReader->suppInfo.numOfCols, pReader->idStr);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return false; return false;
} }
...@@ -2009,12 +2013,12 @@ static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) { ...@@ -2009,12 +2013,12 @@ static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) {
} }
static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; } static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; }
bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) { bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) {
if (pBlockData->nRow > 0) { if (pBlockData->nRow > 0) {
ASSERT(pBlockData->nRow == pDumpInfo->totalRows); ASSERT(pBlockData->nRow == pDumpInfo->totalRows);
} }
return pBlockData->nRow > 0 && (!pDumpInfo->allDumped); return pBlockData->nRow > 0 && (!pDumpInfo->allDumped);
} }
int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key, int32_t mergeRowsInFileBlocks(SBlockData* pBlockData, STableBlockScanInfo* pBlockScanInfo, int64_t key,
...@@ -2452,7 +2456,8 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) { ...@@ -2452,7 +2456,8 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
code = buildComposedDataBlock(pReader); code = buildComposedDataBlock(pReader);
} else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) { } else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) {
tBlockDataReset(&pStatus->fileBlockData); tBlockDataReset(&pStatus->fileBlockData);
code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pScanInfo->uid, pReader->pSchema); TABLEID tid = {.suid = pReader->suid, .uid = pScanInfo->uid};
code = tBlockDataInit(&pStatus->fileBlockData, &tid, pReader->pSchema, NULL, 0);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }
...@@ -2932,7 +2937,8 @@ static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanIn ...@@ -2932,7 +2937,8 @@ static int32_t checkForNeighborFileBlock(STsdbReader* pReader, STableBlockScanIn
// 3. load the neighbor block, and set it to be the currently accessed file data block // 3. load the neighbor block, and set it to be the currently accessed file data block
tBlockDataReset(&pStatus->fileBlockData); tBlockDataReset(&pStatus->fileBlockData);
int32_t code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pFBlock->uid, pReader->pSchema); TABLEID tid = {.suid = pReader->suid, .uid = pFBlock->uid};
int32_t code = tBlockDataInit(&pStatus->fileBlockData, &tid, pReader->pSchema, NULL, 0);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }
...@@ -3404,10 +3410,14 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTabl ...@@ -3404,10 +3410,14 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* pTabl
// we need only one row // we need only one row
pPrevReader->capacity = 1; pPrevReader->capacity = 1;
pPrevReader->status.pTableMap = pReader->status.pTableMap; pPrevReader->status.pTableMap = pReader->status.pTableMap;
pPrevReader->pSchema = pReader->pSchema;
pPrevReader->pMemSchema = pReader->pMemSchema;
pPrevReader->pReadSnap = pReader->pReadSnap; pPrevReader->pReadSnap = pReader->pReadSnap;
pNextReader->capacity = 1; pNextReader->capacity = 1;
pNextReader->status.pTableMap = pReader->status.pTableMap; pNextReader->status.pTableMap = pReader->status.pTableMap;
pNextReader->pSchema = pReader->pSchema;
pNextReader->pMemSchema = pReader->pMemSchema;
pNextReader->pReadSnap = pReader->pReadSnap; pNextReader->pReadSnap = pReader->pReadSnap;
code = doOpenReaderImpl(pPrevReader); code = doOpenReaderImpl(pPrevReader);
...@@ -3441,11 +3451,19 @@ void tsdbReaderClose(STsdbReader* pReader) { ...@@ -3441,11 +3451,19 @@ void tsdbReaderClose(STsdbReader* pReader) {
{ {
if (pReader->innerReader[0] != NULL) { if (pReader->innerReader[0] != NULL) {
pReader->innerReader[0]->status.pTableMap = NULL; STsdbReader* p = pReader->innerReader[0];
pReader->innerReader[0]->pReadSnap = NULL;
pReader->innerReader[1]->status.pTableMap = NULL; p->status.pTableMap = NULL;
pReader->innerReader[1]->pReadSnap = NULL; p->pReadSnap = NULL;
p->pSchema = NULL;
p->pMemSchema = NULL;
p = pReader->innerReader[1];
p->status.pTableMap = NULL;
p->pReadSnap = NULL;
p->pSchema = NULL;
p->pMemSchema = NULL;
tsdbReaderClose(pReader->innerReader[0]); tsdbReaderClose(pReader->innerReader[0]);
tsdbReaderClose(pReader->innerReader[1]); tsdbReaderClose(pReader->innerReader[1]);
...@@ -3684,7 +3702,8 @@ static SArray* doRetrieveDataBlock(STsdbReader* pReader) { ...@@ -3684,7 +3702,8 @@ static SArray* doRetrieveDataBlock(STsdbReader* pReader) {
STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pFBlock->uid, sizeof(pFBlock->uid)); STableBlockScanInfo* pBlockScanInfo = taosHashGet(pStatus->pTableMap, &pFBlock->uid, sizeof(pFBlock->uid));
tBlockDataReset(&pStatus->fileBlockData); tBlockDataReset(&pStatus->fileBlockData);
int32_t code = tBlockDataInit(&pStatus->fileBlockData, pReader->suid, pBlockScanInfo->uid, pReader->pSchema); TABLEID tid = {.suid = pReader->suid, .uid = pBlockScanInfo->uid};
int32_t code = tBlockDataInit(&pStatus->fileBlockData, &tid, pReader->pSchema, NULL, 0);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
terrno = code; terrno = code;
return NULL; return NULL;
......
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
浏览文件 @ e17c65fd
...@@ -1013,12 +1013,13 @@ _err: ...@@ -1013,12 +1013,13 @@ _err:
return code; return code;
} }
static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, SBlockData *pBlockData) { static int32_t tsdbReadBlockDataImpl(SDataFReader *pReader, SBlockInfo *pBlkInfo, SBlockData *pBlockData,
int32_t iStt) {
int32_t code = 0; int32_t code = 0;
tBlockDataClear(pBlockData); tBlockDataClear(pBlockData);
STsdbFD *pFD = pReader->pDataFD; STsdbFD *pFD = (iStt < 0) ? pReader->pDataFD : pReader->aSttFD[iStt];
// uid + version + tskey // uid + version + tskey
code = tRealloc(&pReader->aBuf[0], pBlkInfo->szKey); code = tRealloc(&pReader->aBuf[0], pBlkInfo->szKey);
...@@ -1157,9 +1158,12 @@ _err: ...@@ -1157,9 +1158,12 @@ _err:
int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) { int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData *pBlockData) {
int32_t code = 0; int32_t code = 0;
code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], pBlockData); code = tsdbReadBlockDataImpl(pReader, &pDataBlk->aSubBlock[0], pBlockData, -1);
if (code) goto _err; if (code) goto _err;
ASSERT(pDataBlk->nSubBlock == 1);
#if 0
if (pDataBlk->nSubBlock > 1) { if (pDataBlk->nSubBlock > 1) {
SBlockData bData1; SBlockData bData1;
SBlockData bData2; SBlockData bData2;
...@@ -1200,6 +1204,7 @@ int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData ...@@ -1200,6 +1204,7 @@ int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pDataBlk, SBlockData
tBlockDataDestroy(&bData1, 1); tBlockDataDestroy(&bData1, 1);
tBlockDataDestroy(&bData2, 1); tBlockDataDestroy(&bData2, 1);
} }
#endif
return code; return code;
...@@ -1210,23 +1215,38 @@ _err: ...@@ -1210,23 +1215,38 @@ _err:
int32_t tsdbReadSttBlock(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData) { int32_t tsdbReadSttBlock(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData) {
int32_t code = 0; int32_t code = 0;
int32_t lino = 0;
code = tsdbReadBlockDataImpl(pReader, &pSttBlk->bInfo, pBlockData, iStt);
TSDB_CHECK_CODE(code, lino, _exit);
_exit:
if (code) {
tsdbError("vgId:%d %s failed at %d since %s", TD_VID(pReader->pTsdb->pVnode), __func__, lino, tstrerror(code));
}
return code;
}
int32_t tsdbReadSttBlockEx(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData) {
int32_t code = 0;
int32_t lino = 0;
// alloc // alloc
code = tRealloc(&pReader->aBuf[0], pSttBlk->bInfo.szBlock); code = tRealloc(&pReader->aBuf[0], pSttBlk->bInfo.szBlock);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
// read // read
code = tsdbReadFile(pReader->aSttFD[iStt], pSttBlk->bInfo.offset, pReader->aBuf[0], pSttBlk->bInfo.szBlock); code = tsdbReadFile(pReader->aSttFD[iStt], pSttBlk->bInfo.offset, pReader->aBuf[0], pSttBlk->bInfo.szBlock);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
// decmpr // decmpr
code = tDecmprBlockData(pReader->aBuf[0], pSttBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]); code = tDecmprBlockData(pReader->aBuf[0], pSttBlk->bInfo.szBlock, pBlockData, &pReader->aBuf[1]);
if (code) goto _err; TSDB_CHECK_CODE(code, lino, _exit);
return code;
_err: _exit:
tsdbError("vgId:%d tsdb read stt block failed since %s", TD_VID(pReader->pTsdb->pVnode), tstrerror(code)); if (code) {
tsdbError("vgId:%d %s failed at %d since %s", TD_VID(pReader->pTsdb->pVnode), __func__, lino, tstrerror(code));
}
return code; return code;
} }
......
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
浏览文件 @ e17c65fd
...@@ -140,7 +140,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) { ...@@ -140,7 +140,7 @@ static int32_t tsdbSnapReadOpenFile(STsdbSnapReader* pReader) {
if (pSttBlk->minVer > pReader->ever) continue; if (pSttBlk->minVer > pReader->ever) continue;
if (pSttBlk->maxVer < pReader->sver) continue; if (pSttBlk->maxVer < pReader->sver) continue;
code = tsdbReadSttBlock(pReader->pDataFReader, iStt, pSttBlk, &pIter->bData); code = tsdbReadSttBlockEx(pReader->pDataFReader, iStt, pSttBlk, &pIter->bData);
if (code) goto _err; if (code) goto _err;
for (pIter->iRow = 0; pIter->iRow < pIter->bData.nRow; pIter->iRow++) { for (pIter->iRow = 0; pIter->iRow < pIter->bData.nRow; pIter->iRow++) {
...@@ -223,7 +223,7 @@ static int32_t tsdbSnapNextRow(STsdbSnapReader* pReader) { ...@@ -223,7 +223,7 @@ static int32_t tsdbSnapNextRow(STsdbSnapReader* pReader) {
if (pSttBlk->minVer > pReader->ever || pSttBlk->maxVer < pReader->sver) continue; if (pSttBlk->minVer > pReader->ever || pSttBlk->maxVer < pReader->sver) continue;
code = tsdbReadSttBlock(pReader->pDataFReader, pIter->iStt, pSttBlk, &pIter->bData); code = tsdbReadSttBlockEx(pReader->pDataFReader, pIter->iStt, pSttBlk, &pIter->bData);
if (code) goto _err; if (code) goto _err;
pIter->iRow = -1; pIter->iRow = -1;
...@@ -319,7 +319,7 @@ static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) { ...@@ -319,7 +319,7 @@ static int32_t tsdbSnapReadData(STsdbSnapReader* pReader, uint8_t** ppData) {
code = tsdbUpdateTableSchema(pTsdb->pVnode->pMeta, id.suid, id.uid, &pReader->skmTable); code = tsdbUpdateTableSchema(pTsdb->pVnode->pMeta, id.suid, id.uid, &pReader->skmTable);
if (code) goto _err; if (code) goto _err;
code = tBlockDataInit(pBlockData, id.suid, id.uid, pReader->skmTable.pTSchema); code = tBlockDataInit(pBlockData, &id, pReader->skmTable.pTSchema, NULL, 0);
if (code) goto _err; if (code) goto _err;
while (pRowInfo->suid == id.suid && pRowInfo->uid == id.uid) { while (pRowInfo->suid == id.suid && pRowInfo->uid == id.uid) {
...@@ -715,7 +715,7 @@ static int32_t tsdbSnapWriteTableDataStart(STsdbSnapWriter* pWriter, TABLEID* pI ...@@ -715,7 +715,7 @@ static int32_t tsdbSnapWriteTableDataStart(STsdbSnapWriter* pWriter, TABLEID* pI
if (code) goto _err; if (code) goto _err;
tMapDataReset(&pWriter->dWriter.mDataBlk); tMapDataReset(&pWriter->dWriter.mDataBlk);
code = tBlockDataInit(&pWriter->dWriter.bData, pId->suid, pId->uid, pWriter->skmTable.pTSchema); code = tBlockDataInit(&pWriter->dWriter.bData, pId, pWriter->skmTable.pTSchema, NULL, 0);
if (code) goto _err; if (code) goto _err;
return code; return code;
...@@ -1000,7 +1000,8 @@ static int32_t tsdbSnapWriteToSttFile(STsdbSnapWriter* pWriter, int32_t iRow) { ...@@ -1000,7 +1000,8 @@ static int32_t tsdbSnapWriteToSttFile(STsdbSnapWriter* pWriter, int32_t iRow) {
code = tsdbUpdateTableSchema(pWriter->pTsdb->pVnode->pMeta, pWriter->id.suid, pWriter->id.uid, &pWriter->skmTable); code = tsdbUpdateTableSchema(pWriter->pTsdb->pVnode->pMeta, pWriter->id.suid, pWriter->id.uid, &pWriter->skmTable);
if (code) goto _err; if (code) goto _err;
code = tBlockDataInit(pBData, pWriter->id.suid, pWriter->id.suid ? 0 : pWriter->id.uid, pWriter->skmTable.pTSchema); TABLEID tid = {.suid = pWriter->id.suid, .uid = pWriter->id.suid ? 0 : pWriter->id.uid};
code = tBlockDataInit(pBData, &tid, pWriter->skmTable.pTSchema, NULL, 0);
if (code) goto _err; if (code) goto _err;
} }
......
source/dnode/vnode/src/tsdb/tsdbUtil.c
浏览文件 @ e17c65fd
...@@ -948,24 +948,47 @@ void tBlockDataDestroy(SBlockData *pBlockData, int8_t deepClear) { ...@@ -948,24 +948,47 @@ void tBlockDataDestroy(SBlockData *pBlockData, int8_t deepClear) {
pBlockData->aColData = NULL; pBlockData->aColData = NULL;
} }
int32_t tBlockDataInit(SBlockData *pBlockData, int64_t suid, int64_t uid, STSchema *pTSchema) { int32_t tBlockDataInit(SBlockData *pBlockData, TABLEID *pId, STSchema *pTSchema, int16_t *aCid, int32_t nCid) {
int32_t code = 0; int32_t code = 0;
ASSERT(suid || uid); ASSERT(pId->suid || pId->uid);
pBlockData->suid = suid; pBlockData->suid = pId->suid;
pBlockData->uid = uid; pBlockData->uid = pId->uid;
pBlockData->nRow = 0; pBlockData->nRow = 0;
taosArrayClear(pBlockData->aIdx); taosArrayClear(pBlockData->aIdx);
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) { if (aCid) {
int32_t iColumn = 1;
STColumn *pTColumn = &pTSchema->columns[iColumn]; STColumn *pTColumn = &pTSchema->columns[iColumn];
for (int32_t iCid = 0; iCid < nCid; iCid++) {
while (pTColumn && pTColumn->colId < aCid[iCid]) {
iColumn++;
pTColumn = (iColumn < pTSchema->numOfCols) ? &pTSchema->columns[iColumn] : NULL;
}
SColData *pColData; if (pTColumn == NULL) {
code = tBlockDataAddColData(pBlockData, iColumn - 1, &pColData); break;
if (code) goto _exit; } else if (pTColumn->colId == aCid[iCid]) {
SColData *pColData;
code = tBlockDataAddColData(pBlockData, taosArrayGetSize(pBlockData->aIdx), &pColData);
if (code) goto _exit;
tColDataInit(pColData, pTColumn->colId, pTColumn->type, (pTColumn->flags & COL_SMA_ON) ? 1 : 0);
tColDataInit(pColData, pTColumn->colId, pTColumn->type, (pTColumn->flags & COL_SMA_ON) ? 1 : 0); iColumn++;
pTColumn = (iColumn < pTSchema->numOfCols) ? &pTSchema->columns[iColumn] : NULL;
}
}
} else {
for (int32_t iColumn = 1; iColumn < pTSchema->numOfCols; iColumn++) {
STColumn *pTColumn = &pTSchema->columns[iColumn];
SColData *pColData;
code = tBlockDataAddColData(pBlockData, iColumn - 1, &pColData);
if (code) goto _exit;
tColDataInit(pColData, pTColumn->colId, pTColumn->type, (pTColumn->flags & COL_SMA_ON) ? 1 : 0);
}
} }
_exit: _exit:
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或