Skip to content

T
TDengine

taosdata / TDengine
1 年多 前同步成功

通知 1185
Star 22016
Fork 4786
T
TDengine
提交 64554461 编写于 作者: H Hongze Cheng
浏览文件
操作

more work

上级 9bc64e80
显示空白变更内容
内联 并排
Showing with 178 addition and 117 deletion
source/dnode/vnode/src/tsdb/tsdbCommit.c
浏览文件 @ 64554461
...@@ -398,26 +398,81 @@ _err: ...@@ -398,26 +398,81 @@ _err:
return code; return code;
} }
static int32_t tsdbMergeCommitImpl(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, SBlock *pBlock) { static int32_t tsdbGetOverlapRowNumber(STbDataIter *pIter, SBlock *pBlock) {
int32_t nRow = 0;
TSDBROW *pRow;
TSDBKEY key;
int32_t c = 0;
STbDataIter iter = *pIter;
iter.pRow = NULL;
while (true) {
pRow = tsdbTbDataIterGet(pIter);
if (pRow == NULL) break;
key = tsdbRowKey(pRow);
c = tBlockCmprFn(&(SBlock){.info.maxKey = key, .info.minKey = key}, pBlock);
if (c == 0) {
nRow++;
} else if (c > 0) {
break;
} else {
ASSERT(0);
}
}
return nRow;
}
static int32_t tsdbMergeCommitImpl(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STbDataIter *pIter, SBlock *pBlock,
int8_t toDataOnly) {
int32_t code = 0; int32_t code = 0;
int32_t iRow = 0;
int32_t nRow = 0; int32_t nRow = 0;
int32_t c;
TSDBROW *pRow;
SBlock block = BLOCK_INIT_VAL; SBlock block = BLOCK_INIT_VAL;
TSDBKEY key1;
TSDBKEY key2;
tsdbBlockDataClear(&pCommitter->bDataN);
if (pBlock->last) {
// load last and merge until {pCommitter->maxKey, INT64_MAX} // load last and merge until {pCommitter->maxKey, INT64_MAX}
code = tsdbReadBlockData(pCommitter->pReader, pBlockIdx, pBlock, &pCommitter->bDataO, NULL, 0, NULL, NULL);
if (code) goto _err;
iRow = 0;
nRow = pCommitter->bDataO.nRow;
pRow = tsdbTbDataIterGet(pIter);
while (true) {
if ((pRow == NULL || pRow->pTSRow->ts > pCommitter->maxKey) && (iRow >= nRow)) {
if (pCommitter->bDataN.nRow > 0) {
goto _write_block_data;
} else { } else {
// scan pIter, check how many rows in the block range break;
if (pBlock->nRow + nRow <= pCommitter->maxRow) {
if (pBlock->nSubBlock < TSDB_MAX_SUBBLOCKS) {
// add as a subblock
} else {
// load the block, merge until pBlock->maxKey
} }
} else {
// load the block, merge until pBlock->maxKey
} }
// TODO
_write_block_data:
block.last = pCommitter->bDataN.nRow < pCommitter->minRow ? 1 : 0;
code = tsdbWriteBlockData(pCommitter->pWriter, &pCommitter->bDataN, NULL, pBlockIdx, &block);
if (code) goto _err;
code = tMapDataPutItem(&pCommitter->nBlock, &block, tPutBlock);
if (code) goto _err;
} }
block = BLOCK_INIT_VAL;
tsdbBlockDataClear(&pCommitter->bDataN);
return code;
_err:
tsdbError("vgId:%d merge commit impl failed since %s", TD_VID(pCommitter->pTsdb->pVnode), tstrerror(code));
return code; return code;
} }
...@@ -436,7 +491,7 @@ static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STb ...@@ -436,7 +491,7 @@ static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STb
if (code) goto _err; if (code) goto _err;
} else if (pBlock->last) { } else if (pBlock->last) {
// merge // merge
code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock); code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock, 0);
if (code) goto _err; if (code) goto _err;
} else { } else {
// memory // memory
...@@ -456,9 +511,14 @@ static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STb ...@@ -456,9 +511,14 @@ static int32_t tsdbMergeCommit(SCommitter *pCommitter, SBlockIdx *pBlockIdx, STb
code = tMapDataPutItem(&pCommitter->nBlock, pBlock, tPutBlock); code = tMapDataPutItem(&pCommitter->nBlock, pBlock, tPutBlock);
if (code) goto _err; if (code) goto _err;
} else if (c == 0) { } else if (c == 0) {
// merge int32_t nOverlap = tsdbGetOverlapRowNumber(pIter, pBlock);
code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock);
if (pBlock->nRow + nOverlap > pCommitter->maxRow || pBlock->nSubBlock == TSDB_MAX_SUBBLOCKS) {
code = tsdbMergeCommitImpl(pCommitter, pBlockIdx, pIter, pBlock, 1);
if (code) goto _err; if (code) goto _err;
} else {
// add as a subblock
}
} else { } else {
ASSERT(0); ASSERT(0);
} }
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 64554461
...@@ -113,7 +113,7 @@ struct STsdbReader { ...@@ -113,7 +113,7 @@ struct STsdbReader {
// SColumnDataAgg* statis; // query level statistics, only one table block statistics info exists at any time // 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 // SColumnDataAgg** pstatis;// the ptr array list to return to caller
int32_t numOfBlocks; int32_t numOfBlocks;
SArray* pColumns; // column list, SColumnInfoData array list SArray* pColumns; // SArray<SColumnInfoData>
bool locateStart; bool locateStart;
int32_t outputCapacity; int32_t outputCapacity;
int32_t realNumOfRows; int32_t realNumOfRows;
...@@ -180,42 +180,43 @@ struct STsdbReader { ...@@ -180,42 +180,43 @@ struct STsdbReader {
// return pLocalIdList; // return pLocalIdList;
// } // }
// static SArray* createCheckInfoFromTableGroup(STsdbReader* pTsdbReadHandle, STableListInfo* pTableList) { static SArray* createCheckInfoFromTableGroup(STsdbReader* pTsdbReadHandle, STableListInfo* pTableList) {
// size_t tableSize = taosArrayGetSize(pTableList->pTableList); // size_t tableSize = taosArrayGetSize(pTableList->pTableList);
// assert(tableSize >= 1); // assert(tableSize >= 1);
// // allocate buffer in order to load data blocks from file // // allocate buffer in order to load data blocks from file
// SArray* pTableCheckInfo = taosArrayInit(tableSize, sizeof(STableCheckInfo)); // SArray* pTableCheckInfo = taosArrayInit(tableSize, sizeof(STableCheckInfo));
// if (pTableCheckInfo == NULL) { // if (pTableCheckInfo == NULL) {
// return NULL; // return NULL;
// } // }
// // todo apply the lastkey of table check to avoid to load header file
// for (int32_t j = 0; j < tableSize; ++j) {
// STableKeyInfo* pKeyInfo = (STableKeyInfo*)taosArrayGet(pTableList->pTableList, j);
// STableCheckInfo info = {.lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid}; // // todo apply the lastkey of table check to avoid to load header file
// info.suid = pTsdbReadHandle->suid; // for (int32_t j = 0; j < tableSize; ++j) {
// if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) { // STableKeyInfo* pKeyInfo = (STableKeyInfo*)taosArrayGet(pTableList->pTableList, j);
// if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) {
// info.lastKey = pTsdbReadHandle->window.skey; // STableCheckInfo info = {.lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid};
// } // info.suid = pTsdbReadHandle->suid;
// if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
// if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) {
// info.lastKey = pTsdbReadHandle->window.skey;
// }
// assert(info.lastKey >= pTsdbReadHandle->window.skey && info.lastKey <= pTsdbReadHandle->window.ekey); // assert(info.lastKey >= pTsdbReadHandle->window.skey && info.lastKey <= pTsdbReadHandle->window.ekey);
// } else { // } else {
// info.lastKey = pTsdbReadHandle->window.skey; // info.lastKey = pTsdbReadHandle->window.skey;
// } // }
// taosArrayPush(pTableCheckInfo, &info); // taosArrayPush(pTableCheckInfo, &info);
// tsdbDebug("%p check table uid:%" PRId64 " from lastKey:%" PRId64 " %s", pTsdbReadHandle, info.tableId, // tsdbDebug("%p check table uid:%" PRId64 " from lastKey:%" PRId64 " %s", pTsdbReadHandle, info.tableId,
// info.lastKey, // info.lastKey,
// pTsdbReadHandle->idStr); // pTsdbReadHandle->idStr);
// } // }
// // TODO group table according to the tag value. // // TODO group table according to the tag value.
// taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar); // taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar);
// return pTableCheckInfo; // return pTableCheckInfo;
// } return NULL;
}
// static void resetCheckInfo(STsdbReader* pTsdbReadHandle) { // static void resetCheckInfo(STsdbReader* pTsdbReadHandle) {
// size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo); // size_t numOfTables = taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo);
...@@ -266,31 +267,31 @@ struct STsdbReader { ...@@ -266,31 +267,31 @@ struct STsdbReader {
// return now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1; // needs to add one tick // return now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1; // needs to add one tick
// } // }
// static void setQueryTimewindow(STsdbReader* pTsdbReadHandle, SQueryTableDataCond* pCond, int32_t tWinIdx) { static void setQueryTimewindow(STsdbReader* pReader, SQueryTableDataCond* pCond, int32_t tWinIdx) {
// pTsdbReadHandle->window = pCond->twindows[tWinIdx]; // pReader->window = pCond->twindows[tWinIdx];
// bool updateTs = false; // bool updateTs = false;
// int64_t startTs = getEarliestValidTimestamp(pTsdbReadHandle->pTsdb); // int64_t startTs = getEarliestValidTimestamp(pReader->pTsdb);
// if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) { // if (ASCENDING_TRAVERSE(pReader->order)) {
// if (startTs > pTsdbReadHandle->window.skey) { // if (startTs > pReader->window.skey) {
// pTsdbReadHandle->window.skey = startTs; // pReader->window.skey = startTs;
// pCond->twindows[tWinIdx].skey = startTs; // pCond->twindows[tWinIdx].skey = startTs;
// updateTs = true; // updateTs = true;
// } // }
// } else { // } else {
// if (startTs > pTsdbReadHandle->window.ekey) { // if (startTs > pReader->window.ekey) {
// pTsdbReadHandle->window.ekey = startTs; // pReader->window.ekey = startTs;
// pCond->twindows[tWinIdx].ekey = startTs; // pCond->twindows[tWinIdx].ekey = startTs;
// updateTs = true; // updateTs = true;
// } // }
// } // }
// if (updateTs) { // if (updateTs) {
// tsdbDebug("%p update the query time window, old:%" PRId64 " - %" PRId64 ", new:%" PRId64 " - %" PRId64 ", %s", // tsdbDebug("%p update the query time window, old:%" PRId64 " - %" PRId64 ", new:%" PRId64 " - %" PRId64 ", %s",
// pTsdbReadHandle, pCond->twindows[tWinIdx].skey, pCond->twindows[tWinIdx].ekey, // pReader, pCond->twindows[tWinIdx].skey, pCond->twindows[tWinIdx].ekey, pReader->window.skey,
// pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, pTsdbReadHandle->idStr); // pReader->window.ekey, pReader->idStr);
// } // }
// } }
static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId, static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint64_t qId, uint64_t taskId,
STsdbReader** ppReader) { STsdbReader** ppReader) {
...@@ -326,9 +327,9 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint ...@@ -326,9 +327,9 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint
// // goto _end; // // goto _end;
// // } // // }
// setQueryTimewindow(pReadHandle, pCond, 0); setQueryTimewindow(pReader, pCond, 0);
// if (pCond->numOfCols > 0) { if (pCond->numOfCols > 0) {
// int32_t rowLen = 0; // int32_t rowLen = 0;
// for (int32_t i = 0; i < pCond->numOfCols; ++i) { // for (int32_t i = 0; i < pCond->numOfCols; ++i) {
// rowLen += pCond->colList[i].bytes; // rowLen += pCond->colList[i].bytes;
...@@ -369,7 +370,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint ...@@ -369,7 +370,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, uint
// size_t size = taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn); // size_t size = taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn);
// pReadHandle->suppInfo.slotIds = taosMemoryCalloc(size, sizeof(int32_t)); // pReadHandle->suppInfo.slotIds = taosMemoryCalloc(size, sizeof(int32_t));
// pReadHandle->suppInfo.plist = taosMemoryCalloc(size, POINTER_BYTES); // pReadHandle->suppInfo.plist = taosMemoryCalloc(size, POINTER_BYTES);
// } }
// pReadHandle->pDataCols = tdNewDataCols(1000, pVnode->config.tsdbCfg.maxRows); // pReadHandle->pDataCols = tdNewDataCols(1000, pVnode->config.tsdbCfg.maxRows);
// if (pReadHandle->pDataCols == NULL) { // if (pReadHandle->pDataCols == NULL) {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或