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

TD-100

This commit is contained in:
hzcheng 2020-04-14 09:33:14 +08:00
parent d28fae1a45
commit 66cf4a5bac
2 changed files with 9 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/vnode/tsdb/src/tsdbMain.c
View File

@ -760,6 +760,8 @@ static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int max
int numOfRows = 0; int numOfRows = 0;
do { do {
if (numOfRows >= maxRowsToRead) break;
SSkipListNode *node = tSkipListIterGet(pIter); SSkipListNode *node = tSkipListIterGet(pIter);
if (node == NULL) break; if (node == NULL) break;
@ -769,7 +771,6 @@ static int tsdbReadRowsFromCache(SSkipListIterator *pIter, TSKEY maxKey, int max
tdAppendDataRowToDataCol(row, pCols); tdAppendDataRowToDataCol(row, pCols);
numOfRows++; numOfRows++;
if (numOfRows >= maxRowsToRead) break;
} while (tSkipListIterNext(pIter)); } while (tSkipListIterNext(pIter));
return numOfRows; return numOfRows;

29
src/vnode/tsdb/src/tsdbRWHelper.c
View File

@ -243,27 +243,11 @@ int tsdbWriteDataBlock(SRWHelper *pHelper, SDataCols *pDataCols) {
rowsToWrite = tsdbMergeDataWithBlock(pHelper, blkIdx, pDataCols); rowsToWrite = tsdbMergeDataWithBlock(pHelper, blkIdx, pDataCols);
if (rowsToWrite < 0) goto _err; if (rowsToWrite < 0) goto _err;
} else { // Save as a super block in the middle
} else { // Either merge with the previous block or save as a super block in the middle int rowsToWrite = tsdbGetRowsInRange(pDataCols, 0, pCompBlock->keyFirst-1);
SCompBlock *prevBlock = (blkIdx == 0) ? NULL : (pCompBlock - 1); ASSERT(rowsToWrite > 0);
if (tsdbWriteBlockToFile(pHelper, &(pHelper->files.dataF), pDataCols, rowsToWrite, &compBlock, false, true) < 0) goto _err;
int rows1 = nRowsLEThan(pDataCols, pCompBlock->keyFirst); // rows write as a super block in the middle if (tsdbInsertSuperBlock(pHelper, pCompBlock, pCompBlock - pHelper->pCompInfo->blocks) < 0) goto _err;
int rows2 = (prevBlock) ? (pHelper->config.maxRowsPerFileBlock - prevBlock->numOfPoints)
: rows1; // rows can merge with the previous block
if (rows1 >= rows2) {
rowsToWrite = tsdbWriteBlockToFile(pHelper, &(pHelper->files.dataF), pDataCols, rows1, &compBlock, false, true);
if (rowsToWrite < 0) goto _err;
ASSERT(rowsToWrite == rows1);
// Add the super block to it
pIdx->len += sizeof(SCompBlock);
pIdx->numOfSuperBlocks++;
} else {
rowsToWrite = tsdbMergeDataWithBlock(pHelper, blkIdx-1, pDataCols);
if (rowsToWrite < 0) goto _err;
ASSERT(rowsToWrite == rows2);
}
} }
} }
} }
@ -671,7 +655,7 @@ static int nRowsLEThan(SDataCols *pDataCols, int maxKey) {
return ((TSKEY *)ptr - (TSKEY *)(pDataCols->cols[0].pData)) + 1; return ((TSKEY *)ptr - (TSKEY *)(pDataCols->cols[0].pData)) + 1;
} }
// Merge the data with a block // Merge the data with a block in file
static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDataCols) { static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDataCols) {
// TODO: set pHelper->hasOldBlock // TODO: set pHelper->hasOldBlock
int rowsWritten = 0; int rowsWritten = 0;
@ -685,6 +669,7 @@ static int tsdbMergeDataWithBlock(SRWHelper *pHelper, int blkIdx, SDataCols *pDa
SCompBlock *pCompBlock = pHelper->pCompInfo->blocks + blkIdx; SCompBlock *pCompBlock = pHelper->pCompInfo->blocks + blkIdx;
ASSERT(pCompBlock->numOfSubBlocks >= 1); ASSERT(pCompBlock->numOfSubBlocks >= 1);
ASSERT(keyFirst >= pCompBlock->keyFirst); ASSERT(keyFirst >= pCompBlock->keyFirst);
ASSERT(compareKeyBlock((void *)&keyFirst, (void *)pCompBlock) == 0);
// Start here // Start here
TSKEY keyLimit = TSKEY keyLimit =