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

Merge pull request #26881 from taosdata/fix/create_tb 

fix(tsdb): check return value
This commit is contained in:
Haojun Liao 2024-07-30 19:11:16 +08:00 committed by GitHub
parent 51a28996c0 7d250d0d8e
commit 2c73757aa5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 128 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

161
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View File

@ -301,8 +301,10 @@ static int32_t binarySearchForStartRowIndex(uint64_t *uidList, int32_t num, uint
static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray, SSttBlockLoadInfo *pBlockLoadInfo, static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray, SSttBlockLoadInfo *pBlockLoadInfo,
uint64_t suid) { uint64_t suid) {
void *px = NULL;
int32_t code = TSDB_CODE_SUCCESS;
if (TARRAY2_SIZE(pArray) <= 0) { if (TARRAY2_SIZE(pArray) <= 0) {
return TSDB_CODE_SUCCESS; return code;
} }
SSttBlk *pStart = &pArray->data[0]; SSttBlk *pStart = &pArray->data[0];
@ -316,10 +318,17 @@ static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else { // all blocks are qualified } else { // all blocks are qualified
taosArrayClear(pBlockLoadInfo->aSttBlk); taosArrayClear(pBlockLoadInfo->aSttBlk);
taosArrayAddBatch(pBlockLoadInfo->aSttBlk, pArray->data, pArray->size); px = taosArrayAddBatch(pBlockLoadInfo->aSttBlk, pArray->data, pArray->size);
if (px == NULL){
return terrno;
}
} }
} else { } else {
SArray *pTmp = taosArrayInit(TARRAY2_SIZE(pArray), sizeof(SSttBlk)); SArray *pTmp = taosArrayInit(TARRAY2_SIZE(pArray), sizeof(SSttBlk));
if (pTmp == NULL) {
return terrno;
}
for (int32_t i = 0; i < TARRAY2_SIZE(pArray); ++i) { for (int32_t i = 0; i < TARRAY2_SIZE(pArray); ++i) {
SSttBlk *p = &pArray->data[i]; SSttBlk *p = &pArray->data[i];
if (p->suid < suid) { if (p->suid < suid) {
@ -327,7 +336,11 @@ static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray
} }
if (p->suid == suid) { if (p->suid == suid) {
taosArrayPush(pTmp, p); void* px = taosArrayPush(pTmp, p);
if (px == NULL) {
code = terrno;
break;
}
} else if (p->suid > suid) { } else if (p->suid > suid) {
break; break;
} }
@ -337,7 +350,7 @@ static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray
pBlockLoadInfo->aSttBlk = pTmp; pBlockLoadInfo->aSttBlk = pTmp;
} }
return TSDB_CODE_SUCCESS; return code;
} }
static int32_t tValueDupPayload(SValue *pVal) { static int32_t tValueDupPayload(SValue *pVal) {
@ -357,9 +370,11 @@ static int32_t tValueDupPayload(SValue *pVal) {
static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBlockLoadInfo *pBlockLoadInfo, static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBlockLoadInfo *pBlockLoadInfo,
TStatisBlkArray *pStatisBlkArray, uint64_t suid, const char *id) { TStatisBlkArray *pStatisBlkArray, uint64_t suid, const char *id) {
int32_t code = TSDB_CODE_SUCCESS;
void* px = NULL;
int32_t numOfBlocks = TARRAY2_SIZE(pStatisBlkArray); int32_t numOfBlocks = TARRAY2_SIZE(pStatisBlkArray);
if (numOfBlocks <= 0) { if (numOfBlocks <= 0) {
return 0; return code;
} }
int32_t startIndex = 0; int32_t startIndex = 0;
@ -385,7 +400,10 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
for (int32_t k = startIndex; k < endIndex; ++k) { for (int32_t k = startIndex; k < endIndex; ++k) {
tsdbSttFileReadStatisBlock(pSttFileReader, &pStatisBlkArray->data[k], &block); code = tsdbSttFileReadStatisBlock(pSttFileReader, &pStatisBlkArray->data[k], &block);
if (code) {
return code;
}
int32_t i = 0; int32_t i = 0;
int32_t rows = block.numOfRecords; int32_t rows = block.numOfRecords;
@ -409,21 +427,43 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
int32_t size = rows - i; int32_t size = rows - i;
int32_t offset = i * sizeof(int64_t); int32_t offset = i * sizeof(int64_t);
taosArrayAddBatch(pBlockLoadInfo->info.pUid, tBufferGetDataAt(&block.uids, offset), size); px = taosArrayAddBatch(pBlockLoadInfo->info.pUid, tBufferGetDataAt(&block.uids, offset), size);
taosArrayAddBatch(pBlockLoadInfo->info.pFirstTs, tBufferGetDataAt(&block.firstKeyTimestamps, offset), size); if (px == NULL) {
taosArrayAddBatch(pBlockLoadInfo->info.pLastTs, tBufferGetDataAt(&block.lastKeyTimestamps, offset), size); return terrno;
taosArrayAddBatch(pBlockLoadInfo->info.pCount, tBufferGetDataAt(&block.counts, offset), size); }
px = taosArrayAddBatch(pBlockLoadInfo->info.pFirstTs, tBufferGetDataAt(&block.firstKeyTimestamps, offset), size);
if (px == NULL){
return terrno;
}
px = taosArrayAddBatch(pBlockLoadInfo->info.pLastTs, tBufferGetDataAt(&block.lastKeyTimestamps, offset), size);
if (px == NULL){
return terrno;
}
px = taosArrayAddBatch(pBlockLoadInfo->info.pCount, tBufferGetDataAt(&block.counts, offset), size);
if (px == NULL){
return terrno;
}
if (block.numOfPKs > 0) { if (block.numOfPKs > 0) {
SValue vFirst = {0}, vLast = {0}; SValue vFirst = {0}, vLast = {0};
for (int32_t f = i; f < rows; ++f) { for (int32_t f = i; f < rows; ++f) {
int32_t code = tValueColumnGet(&block.firstKeyPKs[0], f, &vFirst); code = tValueColumnGet(&block.firstKeyPKs[0], f, &vFirst);
if (code) { if (code) {
break; break;
} }
tValueDupPayload(&vFirst); code = tValueDupPayload(&vFirst);
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &vFirst); if (code) {
break;
}
px = taosArrayPush(pBlockLoadInfo->info.pFirstKey, &vFirst);
if (px == NULL) {
return terrno;
}
// todo add api to clone the original data // todo add api to clone the original data
code = tValueColumnGet(&block.lastKeyPKs[0], f, &vLast); code = tValueColumnGet(&block.lastKeyPKs[0], f, &vLast);
@ -431,14 +471,28 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
break; break;
} }
tValueDupPayload(&vLast); code = tValueDupPayload(&vLast);
taosArrayPush(pBlockLoadInfo->info.pLastKey, &vLast); if (code) {
break;
}
px = taosArrayPush(pBlockLoadInfo->info.pLastKey, &vLast);
if (px == NULL) {
return terrno;
}
} }
} else { } else {
SValue vFirst = {0}; SValue vFirst = {0};
for (int32_t j = 0; j < size; ++j) { for (int32_t j = 0; j < size; ++j) {
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &vFirst); px = taosArrayPush(pBlockLoadInfo->info.pFirstKey, &vFirst);
taosArrayPush(pBlockLoadInfo->info.pLastKey, &vFirst); if (px == NULL) {
return terrno;
}
px = taosArrayPush(pBlockLoadInfo->info.pLastKey, &vFirst);
if (px == NULL) {
return terrno;
}
} }
} }
} else { } else {
@ -450,24 +504,59 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
break; break;
} }
taosArrayPush(pBlockLoadInfo->info.pUid, &record.uid); px = taosArrayPush(pBlockLoadInfo->info.pUid, &record.uid);
taosArrayPush(pBlockLoadInfo->info.pCount, &record.count); if (px == NULL) {
return terrno;
}
taosArrayPush(pBlockLoadInfo->info.pFirstTs, &record.firstKey.ts); px = taosArrayPush(pBlockLoadInfo->info.pCount, &record.count);
taosArrayPush(pBlockLoadInfo->info.pLastTs, &record.lastKey.ts); if (px == NULL) {
return terrno;
}
px = taosArrayPush(pBlockLoadInfo->info.pFirstTs, &record.firstKey.ts);
if (px == NULL) {
return terrno;
}
px = taosArrayPush(pBlockLoadInfo->info.pLastTs, &record.lastKey.ts);
if (px == NULL) {
return terrno;
}
if (record.firstKey.numOfPKs > 0) { if (record.firstKey.numOfPKs > 0) {
SValue s = record.firstKey.pks[0]; SValue s = record.firstKey.pks[0];
tValueDupPayload(&s); code = tValueDupPayload(&s);
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &s); if (code) {
return code;
}
px = taosArrayPush(pBlockLoadInfo->info.pFirstKey, &s);
if (px == NULL) {
return terrno;
}
s = record.lastKey.pks[0]; s = record.lastKey.pks[0];
tValueDupPayload(&s); code = tValueDupPayload(&s);
taosArrayPush(pBlockLoadInfo->info.pLastKey, &s); if (code) {
return code;
}
px = taosArrayPush(pBlockLoadInfo->info.pLastKey, &s);
if (px == NULL) {
return terrno;
}
} else { } else {
SValue v = {0}; SValue v = {0};
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &v); px = taosArrayPush(pBlockLoadInfo->info.pFirstKey, &v);
taosArrayPush(pBlockLoadInfo->info.pLastKey, &v); if (px == NULL) {
return terrno;
}
px = taosArrayPush(pBlockLoadInfo->info.pLastKey, &v);
if (px == NULL) {
return terrno;
}
} }
i += 1; i += 1;
@ -482,7 +571,7 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
pBlockLoadInfo->cost.statisElapsedTime += el; pBlockLoadInfo->cost.statisElapsedTime += el;
tsdbDebug("%s load %d statis blocks into buf, elapsed time:%.2fms", id, num, el); tsdbDebug("%s load %d statis blocks into buf, elapsed time:%.2fms", id, num, el);
return TSDB_CODE_SUCCESS; return code;
} }
static int32_t doLoadSttFilesBlk(SSttBlockLoadInfo *pBlockLoadInfo, SLDataIter *pIter, int64_t suid, static int32_t doLoadSttFilesBlk(SSttBlockLoadInfo *pBlockLoadInfo, SLDataIter *pIter, int64_t suid,
@ -617,7 +706,7 @@ int32_t tLDataIterOpen2(SLDataIter *pIter, SSttFileReader *pSttFileReader, int32
} }
void tLDataIterClose2(SLDataIter *pIter) { void tLDataIterClose2(SLDataIter *pIter) {
tsdbSttFileReaderClose(&pIter->pReader); (void) tsdbSttFileReaderClose(&pIter->pReader); // always return 0
pIter->pReader = NULL; pIter->pReader = NULL;
} }
@ -890,7 +979,10 @@ int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoF
goto _end; goto _end;
} }
adjustSttDataIters(pConf->pSttFileBlockIterArray, pConf->pCurrentFileset); code = adjustSttDataIters(pConf->pSttFileBlockIterArray, pConf->pCurrentFileset);
if (code) {
goto _end;
}
for (int32_t j = 0; j < numOfLevels; ++j) { for (int32_t j = 0; j < numOfLevels; ++j) {
SSttLvl *pSttLevel = ((STFileSet *)pConf->pCurrentFileset)->lvlArr->data[j]; SSttLvl *pSttLevel = ((STFileSet *)pConf->pCurrentFileset)->lvlArr->data[j];
@ -940,7 +1032,10 @@ int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoF
// let's record the time window for current table of uid in the stt files // let's record the time window for current table of uid in the stt files
if (pSttDataInfo != NULL && numOfRows > 0) { if (pSttDataInfo != NULL && numOfRows > 0) {
taosArrayPush(pSttDataInfo->pKeyRangeList, &range); void* px = taosArrayPush(pSttDataInfo->pKeyRangeList, &range);
if (px == NULL) {
return terrno;
}
pSttDataInfo->numOfRows += numOfRows; pSttDataInfo->numOfRows += numOfRows;
} }
} else { } else {
@ -958,7 +1053,7 @@ _end:
return code; return code;
} }
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) { tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pIter); } void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) { (void) tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pIter); }
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree) { return pMTree->ignoreEarlierTs; } bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree) { return pMTree->ignoreEarlierTs; }
@ -1035,7 +1130,7 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
if (pMTree->pIter && pIter) { if (pMTree->pIter && pIter) {
int32_t c = pMTree->rbt.cmprFn(&pMTree->pIter->node, &pIter->node); int32_t c = pMTree->rbt.cmprFn(&pMTree->pIter->node, &pIter->node);
if (c > 0) { if (c > 0) {
tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter); (void) tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
pMTree->pIter = NULL; pMTree->pIter = NULL;
} else { } else {
ASSERT(c); ASSERT(c);