xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

more work

This commit is contained in:
Hongze Cheng 2022-06-16 12:53:25 +00:00
parent 338383182d
commit 75816b7e7a
3 changed files with 183 additions and 232 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

123
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -346,19 +346,31 @@ _err:
return code;
}
static FORCE_INLINE bool tsdbCommitIterEnd(SCommitter *pCommitter, STbDataIter *pIter) {
TSDBROW *pRow = tsdbTbDataIterGet(pIter);
return ((pRow == NULL) || (pRow->pTSRow->ts <= pCommitter->maxKey));
#define ROW_END(pRow, maxKey) (((pRow) == NULL) || ((pRow)->pTSRow->ts > (maxKey)))
static int32_t tsdbMergeCommit(SCommitter *pCommitter, STbDataIter *pIter, SBlock *pBlock) {
int32_t code = 0;
// TODO
return code;
}
static int32_t tsdbCommitTableDataImpl(SCommitter *pCommitter, STbDataIter *pIter, SBlockIdx *pBlockIdx) {
static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBlockIdx *pBlockIdx) {
int32_t code = 0;
int32_t c;
int32_t iBlock;
int32_t nBlock;
SBlock *pBlock;
SBlock block;
SBlockIdx blockIdx; // todo
STbDataIter iter;
STbDataIter *pIter = &iter;
TSDBROW *pRow;
SBlockIdx blockIdx; // TODO
// create iter
if (pTbData) {
tsdbTbDataIterOpen(pTbData, &(TSDBKEY){.ts = pCommitter->minKey, .version = 0}, 0, pIter);
} else {
pIter = NULL;
}
// check
pRow = tsdbTbDataIterGet(pIter);
if (ROW_END(pRow, pCommitter->maxKey) && pBlockIdx == NULL) goto _exit;
// start ================================
tMapDataReset(&pCommitter->oBlock);
@ -369,48 +381,39 @@ static int32_t tsdbCommitTableDataImpl(SCommitter *pCommitter, STbDataIter *pIte
}
// impl ===============================
iBlock = 0;
nBlock = pCommitter->oBlock.nItem;
SBlock block;
SBlock *pBlock = &block;
int32_t iBlock = 0;
int32_t nBlock = pCommitter->oBlock.nItem;
if (iBlock < nBlock) {
pBlock = &block;
tMapDataGetItemByIdx(&pCommitter->nBlock, iBlock, pBlock, tGetBlock);
} else {
pBlock = NULL;
// merge
pRow = tsdbTbDataIterGet(pIter);
while (!ROW_END(pRow, pCommitter->maxKey) && iBlock < nBlock) {
tMapDataGetItemByIdx(&pCommitter->oBlock, iBlock, pBlock, tGetBlock);
code = tsdbMergeCommit(pCommitter, pIter, pBlock);
if (code) goto _err;
pRow = tsdbTbDataIterGet(pIter);
iBlock++;
}
while (true) {
TSDBROW *pRow = tsdbTbDataIterGet(pIter);
bool iterEnd = ((pRow == NULL) || (pRow->pTSRow->ts > pCommitter->maxKey));
bool blockEnd = (pBlock == NULL);
if (iterEnd && blockEnd) break;
// mem
pRow = tsdbTbDataIterGet(pIter);
while (!ROW_END(pRow, pCommitter->maxKey)) {
code = tsdbMergeCommit(pCommitter, pIter, NULL);
if (code) goto _err;
if (!iterEnd && !blockEnd) {
c = tBlockCmprFn(&(SBlock){.maxKey.ts = pRow->pTSRow->ts}, pBlock);
if (c == 0) {
// merge until pBlock->maxKey
// if (pBlock->last), merge until pCommitter->maxKey
// else merge until pBlock->maxKey
} else if (c < 0) {
// tsdbCommitTableMemData(pIter, pBlock);
// if (pBlock->last), merge until pCommitter->maxKey
// else, commit until pBlock->minKey-1
} else {
// tsdbCommitTableDiskData(pBlock);
// if (pBlock->last), merge until pCommitter->maxKey
// else, move the block to new one
}
} else if (!iterEnd) {
// no block on disk, commit to last when there are no enough data
// commit memory data to pCommitter->maxKey
// tsdbCommitTableMemData(pIter, NULL);
} else {
// tsdbCommitTableDiskData(pBlock); // only left block
// if (last block ? ) else ?
// tMapDataPutItem(&pCommitter->nBlock, pBlock, tPutBlock);
iBlock++; // get new SBlock
pRow = tsdbTbDataIterGet(pIter);
}
// disk
while (iBlock < nBlock) {
tMapDataGetItemByIdx(&pCommitter->oBlock, iBlock, pBlock, tGetBlock);
code = tsdbMergeCommit(pCommitter, NULL, pBlock);
if (code) goto _err;
iBlock++;
}
// end ===============================
@ -420,32 +423,6 @@ static int32_t tsdbCommitTableDataImpl(SCommitter *pCommitter, STbDataIter *pIte
code = tMapDataPutItem(&pCommitter->nBlockIdx, &blockIdx, tPutBlockIdx);
if (code) goto _err;
return code;
_err:
tsdbError("vgId:%d commit table data impl failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
return code;
}
static int32_t tsdbCommitTableData(SCommitter *pCommitter, STbData *pTbData, SBlockIdx *pBlockIdx) {
int32_t code = 0;
STbDataIter *pIter = NULL;
STbDataIter iter;
TSDBROW *pRow;
// create iter if can
if (pTbData) {
pIter = &iter;
tsdbTbDataIterOpen(pTbData, &(TSDBKEY){.ts = pCommitter->minKey, .version = 0}, 0, pIter);
}
// check
if (tsdbCommitIterEnd(pCommitter, pIter) && pBlockIdx == NULL) goto _exit;
// impl
code = tsdbCommitTableDataImpl(pCommitter, pIter, pBlockIdx);
if (code) goto _err;
_exit:
pRow = tsdbTbDataIterGet(pIter);
if (pRow) {

106
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -107,8 +107,8 @@ typedef struct SBlockLoadSuppInfo {
struct STsdbReader {
STsdb* pTsdb;
uint64_t suid;
SQueryFilePos cur; // current position
int16_t order;
SQueryFilePos cur; // current position
STimeWindow window; // the primary query time window that applies to all queries
// SColumnDataAgg* statis; // query level statistics, only one table block statistics info exists at any time
// SColumnDataAgg** pstatis;// the ptr array list to return to caller
@ -292,66 +292,31 @@ struct STsdbReader {
// }
// }
// static STsdb* getTsdbByRetentions(SVnode* pVnode, STsdbReader* pReadHandle, TSKEY winSKey, SRetention* retentions) {
// if (VND_IS_RSMA(pVnode)) {
// int level = 0;
// int64_t now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);
static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId,
STsdbReader** ppReader) {
int32_t code = 0;
STsdbReader* pReader = NULL;
// for (int i = 0; i < TSDB_RETENTION_MAX; ++i) {
// SRetention* pRetention = retentions + level;
// if (pRetention->keep <= 0) {
// if (level > 0) {
// --level;
// }
// break;
// }
// if ((now - pRetention->keep) <= winSKey) {
// break;
// }
// ++level;
// }
// if (level == TSDB_RETENTION_L0) {
// tsdbDebug("vgId:%d, read handle %p rsma level %d is selected to query", TD_VID(pVnode), pReadHandle,
// TSDB_RETENTION_L0);
// return VND_RSMA0(pVnode);
// } else if (level == TSDB_RETENTION_L1) {
// tsdbDebug("vgId:%d, read handle %p rsma level %d is selected to query", TD_VID(pVnode), pReadHandle,
// TSDB_RETENTION_L1);
// return VND_RSMA1(pVnode);
// } else {
// tsdbDebug("vgId:%d, read handle %p rsma level %d is selected to query", TD_VID(pVnode), pReadHandle,
// TSDB_RETENTION_L2);
// return VND_RSMA2(pVnode);
// }
// }
// return VND_TSDB(pVnode);
// }
// static STsdbReader* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId) {
// STsdbReader* pReadHandle = taosMemoryCalloc(1, sizeof(STsdbReader));
// if (pReadHandle == NULL) {
// goto _end;
// }
// STsdb* pTsdb = getTsdbByRetentions(pVnode, pReadHandle, pCond->twindows[0].skey,
// pVnode->config.tsdbCfg.retentions);
// pReadHandle->pTsdb = pTsdb;
// pReadHandle->suid = pCond->suid;
// pReadHandle->order = pCond->order;
// pReadHandle->loadType = pCond->type;
// pReadHandle->loadExternalRow = pCond->loadExternalRows;
// pReadHandle->currentLoadExternalRows = pCond->loadExternalRows;
// pReadHandle->type = TSDB_QUERY_TYPE_ALL;
// pReadHandle->cur.fid = INT32_MIN;
// pReadHandle->cur.win = TSWINDOW_INITIALIZER;
// pReadHandle->checkFiles = true;
// pReadHandle->activeIndex = 0; // current active table index
// pReadHandle->allocSize = 0;
// pReadHandle->locateStart = false;
// pReadHandle->outputCapacity = 4096; //((STsdb*)tsdb)->config.maxRowsPerFileBlock;
// alloc
pReader = (STsdbReader*)taosMemoryCalloc(1, sizeof(*pReader));
if (pReader == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
goto _err;
}
pReader->pTsdb = pVnode->pTsdb; // TODO: pass in pTsdb directly
pReader->suid = pCond->suid;
pReader->order = pCond->order;
pReader->loadType = pCond->type;
pReader->loadExternalRow = pCond->loadExternalRows;
pReader->currentLoadExternalRows = pCond->loadExternalRows;
pReader->type = TSDB_QUERY_TYPE_ALL;
pReader->cur.fid = INT32_MIN;
pReader->cur.win = TSWINDOW_INITIALIZER;
pReader->checkFiles = true;
pReader->activeIndex = 0; // current active table index
pReader->allocSize = 0;
pReader->locateStart = false;
pReader->outputCapacity = 4096; //((STsdb*)tsdb)->config.maxRowsPerFileBlock;
// char buf[128] = {0};
// snprintf(buf, tListLen(buf), "TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, qId);
@ -361,7 +326,6 @@ struct STsdbReader {
// // goto _end;
// // }
// assert(pCond != NULL);
// setQueryTimewindow(pReadHandle, pCond, 0);
// if (pCond->numOfCols > 0) {
@ -423,7 +387,15 @@ struct STsdbReader {
// tsdbReaderClose(pReadHandle);
// terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
// return NULL;
// }
*ppReader = pReader;
return code;
_err:
// tsdbError("");
*ppReader = NULL;
return code;
}
// static int32_t setCurrentSchema(SVnode* pVnode, STsdbReader* pTsdbReadHandle) {
// STableCheckInfo* pCheckInfo = taosArrayGet(pTsdbReadHandle->pTableCheckInfo, 0);
@ -2659,10 +2631,8 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, STableListInf
uint64_t taskId, STsdbReader** ppReader) {
int32_t code = 0;
// STsdbReader* pReader = tsdbQueryTablesImpl(pVnode, pCond, qId, taskId);
// if (pReader == NULL) {
// return NULL;
// }
code = tsdbReaderCreate(pVnode, pCond, qId, taskId, ppReader);
if (code) goto _err;
// if (emptyQueryTimewindow(pReader)) {
// return (STsdbReader*)pReader;
@ -2706,6 +2676,10 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, STableListInf
// taosArrayGetSize(pReader->pTableCheckInfo), taosArrayGetSize(tableList->pTableList), pReader->idStr);
return code;
_err:
// tsdbError("");
return code;
}
void tsdbReaderClose(STsdbReader* pReader) {