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

refactor: do some internal refactor.

This commit is contained in:
Haojun Liao 2024-03-22 17:31:44 +08:00
parent cbf22ed72a
commit 5808c1b86f
1 changed files with 54 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

168
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -1697,7 +1697,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
// be changed, resulting in the corresponding changing of the value of sttRowKey // be changed, resulting in the corresponding changing of the value of sttRowKey
tRowKeyAssign(&pBlockScanInfo->lastProcKey, &minKey); tRowKeyAssign(&pBlockScanInfo->lastProcKey, &minKey);
// ASC: file block ---> stt block -----> mem // file block ---> stt block -----> mem
if (pkCompEx(compFn, &minKey, pfKey) == 0) { if (pkCompEx(compFn, &minKey, pfKey) == 0) {
int32_t code = tsdbRowMergerAdd(pMerger, &fRow, NULL); int32_t code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -1772,7 +1772,7 @@ static int32_t mergeFileBlockAndSttBlock(STsdbReader* pReader, SSttBlockReader*
return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pKey, pReader); return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pKey, pReader);
} else { } else {
// row in both stt file blocks and data file blocks // row in both stt file blocks and data file blocks
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex); TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
SRowKey* pSttKey = getCurrentKeyInSttBlock(pSttBlockReader); SRowKey* pSttKey = getCurrentKeyInSttBlock(pSttBlockReader);
int32_t ret = pkCompEx(pSttBlockReader->pkComparFn, pKey, pSttKey); int32_t ret = pkCompEx(pSttBlockReader->pkComparFn, pKey, pSttKey);
@ -1791,42 +1791,27 @@ static int32_t mergeFileBlockAndSttBlock(STsdbReader* pReader, SSttBlockReader*
} }
} }
tRowKeyAssign(&pBlockScanInfo->lastProcKey, pKey);
// the following for key == sttKey->key.ts // the following for key == sttKey->key.ts
// ASC: file block ------> stt block // file block ------> stt block
// DESC: stt block ------> file block
SRow* pTSRow = NULL; SRow* pTSRow = NULL;
if (ASCENDING_TRAVERSE(pReader->info.order)) { code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
code = tsdbRowMergerAdd(pMerger, &fRow, NULL); if (code != TSDB_CODE_SUCCESS) {
if (code != TSDB_CODE_SUCCESS) { return code;
return code;
}
doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pKey, pReader);
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
code = tsdbRowMergerAdd(pMerger, pRow1, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, pSttKey, pMerger, pkSrcSlot, &pReader->info.verRange, pReader->idStr);
} else {
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
code = tsdbRowMergerAdd(pMerger, pRow1, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, pSttKey, pMerger, pkSrcSlot, &pReader->info.verRange, pReader->idStr);
code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pKey, pReader);
} }
doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pKey, pReader);
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
code = tsdbRowMergerAdd(pMerger, pRow1, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, pSttKey, pMerger, pkSrcSlot, &pReader->info.verRange,
pReader->idStr);
code = tsdbRowMergerGetRow(pMerger, &pTSRow); code = tsdbRowMergerGetRow(pMerger, &pTSRow);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
@ -1896,9 +1881,9 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
SRowKey minKey; SRowKey minKey;
if (ASCENDING_TRAVERSE(pReader->info.order)) { if (ASCENDING_TRAVERSE(pReader->info.order)) {
minKey = k; // let's find the minimum minKey = k; // let's find the minimum
if (pkCompEx(compFn, &ik, &minKey) < 0) {//minKey > ik.key.ts) { if (pkCompEx(compFn, &ik, &minKey) < 0) { // minKey > ik.key.ts) {
minKey = ik; minKey = ik;
} }
@ -1910,7 +1895,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
minKey = *pSttKey; minKey = *pSttKey;
} }
} else { } else {
minKey = k; // let find the maximum ts value minKey = k; // let find the maximum ts value
if (pkCompEx(compFn, &ik, &minKey) > 0) { if (pkCompEx(compFn, &ik, &minKey) > 0) {
minKey = ik; minKey = ik;
} }
@ -1926,95 +1911,50 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
tRowKeyAssign(&pBlockScanInfo->lastProcKey, &minKey); tRowKeyAssign(&pBlockScanInfo->lastProcKey, &minKey);
// ASC: file block -----> stt block -----> imem -----> mem // file block -----> stt block -----> imem -----> mem
// DESC: mem -----> imem -----> stt block -----> file block // if (ASCENDING_TRAVERSE(pReader->info.order)) {
if (ASCENDING_TRAVERSE(pReader->info.order)) { if (pkCompEx(compFn, &minKey, pfKey) == 0) {
if (pkCompEx(compFn, &minKey, pfKey) == 0) { TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex); code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
code = tsdbRowMergerAdd(pMerger, &fRow, NULL); if (code != TSDB_CODE_SUCCESS) {
if (code != TSDB_CODE_SUCCESS) { return code;
return code;
}
doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pfKey, pReader);
} }
if (pkCompEx(compFn, &minKey, pSttKey) == 0) { doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pfKey, pReader);
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree); }
code = tsdbRowMergerAdd(pMerger, pRow1, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, pSttKey, pMerger, pkSrcSlot, &pReader->info.verRange, pReader->idStr); if (pkCompEx(compFn, &minKey, pSttKey) == 0) {
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
code = tsdbRowMergerAdd(pMerger, pRow1, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
} }
if (pkCompEx(compFn, &minKey, &ik) == 0) { doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, pSttKey, pMerger, pkSrcSlot, &pReader->info.verRange,
code = tsdbRowMergerAdd(pMerger, piRow, piSchema); pReader->idStr);
if (code != TSDB_CODE_SUCCESS) { }
return code;
}
code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, &ik, pBlockScanInfo->delSkyline, pReader); if (pkCompEx(compFn, &minKey, &ik) == 0) {
if (code != TSDB_CODE_SUCCESS) { code = tsdbRowMergerAdd(pMerger, piRow, piSchema);
return code; if (code != TSDB_CODE_SUCCESS) {
} return code;
} }
if (pkCompEx(compFn, &minKey, &k) == 0) { code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, &ik, pBlockScanInfo->delSkyline, pReader);
code = tsdbRowMergerAdd(pMerger, pRow, pSchema); if (code != TSDB_CODE_SUCCESS) {
if (code != TSDB_CODE_SUCCESS) { return code;
return code;
}
code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, &k, pBlockScanInfo->delSkyline, pReader);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
} }
} else { }
if (pkCompEx(compFn, &minKey, &k) == 0) {
code = tsdbRowMergerAdd(pMerger, pRow, pSchema);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, &k, pBlockScanInfo->delSkyline, pReader); if (pkCompEx(compFn, &minKey, &k) == 0) {
if (code != TSDB_CODE_SUCCESS) { code = tsdbRowMergerAdd(pMerger, pRow, pSchema);
return code; if (code != TSDB_CODE_SUCCESS) {
} return code;
} }
if (pkCompEx(compFn, &minKey, &ik) == 0) { code = doMergeRowsInBuf(&pBlockScanInfo->iter, pBlockScanInfo->uid, &k, pBlockScanInfo->delSkyline, pReader);
code = tsdbRowMergerAdd(pMerger, piRow, piSchema); if (code != TSDB_CODE_SUCCESS) {
if (code != TSDB_CODE_SUCCESS) { return code;
return code;
}
code = doMergeRowsInBuf(&pBlockScanInfo->iiter, pBlockScanInfo->uid, &ik, pBlockScanInfo->delSkyline, pReader);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
}
if (pkCompEx(compFn, &minKey, pSttKey) == 0) {
TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
code = tsdbRowMergerAdd(pMerger, pRow1, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, pSttKey, pMerger, pkSrcSlot, &pReader->info.verRange, pReader->idStr);
}
if (pkCompEx(compFn, &minKey, pfKey) == 0) {
TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pfKey, pReader);
} }
} }