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

Merge pull request #27289 from taosdata/fix/3_liaohj 

fix(query): return error code.
This commit is contained in:
Haojun Liao 2024-08-20 10:59:07 +08:00 committed by GitHub
parent 21d7820bb5 051763e71f
commit 17c941fc32
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 85 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/dnode/vnode/src/inc/tsdb.h
View File

@ -898,7 +898,7 @@ typedef struct SSttDataInfoForTable {
int32_t tMergeTreeOpen2(SMergeTree *pMTree, SMergeTreeConf *pConf, SSttDataInfoForTable *pTableInfo);
void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
int32_t tMergeTreeNext(SMergeTree *pMTree, bool* pHasNext);
void tMergeTreePinSttBlock(SMergeTree *pMTree);
void tMergeTreeUnpinSttBlock(SMergeTree *pMTree);
bool tMergeTreeIgnoreEarlierTs(SMergeTree *pMTree);

11
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -2245,17 +2245,20 @@ static int32_t lastIterClose(SFSLastIter **iter) {
}
static int32_t lastIterNext(SFSLastIter *iter, TSDBROW **ppRow) {
int32_t code = 0;
bool hasVal = false;
*ppRow = NULL;
int32_t code = tMergeTreeNext(iter->pMergeTree, &hasVal);
if (code != 0) {
return code;
}
bool hasVal = tMergeTreeNext(iter->pMergeTree);
if (!hasVal) {
*ppRow = NULL;
TAOS_RETURN(code);
}
*ppRow = tMergeTreeGetRow(iter->pMergeTree);
TAOS_RETURN(code);
}

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

@ -1100,13 +1100,22 @@ void tMergeTreeUnpinSttBlock(SMergeTree *pMTree) {
tLDataIterUnpinSttBlock(pIter, pMTree->idStr);
}
bool tMergeTreeNext(SMergeTree *pMTree) {
int32_t tMergeTreeNext(SMergeTree *pMTree, bool *pHasNext) {
int32_t code = 0;
if (pHasNext == NULL) {
return TSDB_CODE_INVALID_PARA;
}
*pHasNext = false;
if (pMTree->pIter) {
SLDataIter *pIter = pMTree->pIter;
bool hasVal = false;
int32_t code = tLDataIterNextRow(pIter, pMTree->idStr, &hasVal);
bool hasVal = false;
code = tLDataIterNextRow(pIter, pMTree->idStr, &hasVal);
if (!hasVal || (code != 0)) {
if (code == TSDB_CODE_FILE_CORRUPTED) {
code = 0; // suppress the file corrupt error to enable all queries within this cluster can run without failed.
}
pMTree->pIter = NULL;
}
@ -1115,7 +1124,7 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
if (pMTree->pIter && pIter) {
int32_t c = pMTree->rbt.cmprFn(&pMTree->pIter->node, &pIter->node);
if (c > 0) {
(void) tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
(void)tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
pMTree->pIter = NULL;
} else {
ASSERT(c);
@ -1130,7 +1139,8 @@ bool tMergeTreeNext(SMergeTree *pMTree) {
}
}
return pMTree->pIter != NULL;
*pHasNext = (pMTree->pIter != NULL);
return code;
}
void tMergeTreeClose(SMergeTree *pMTree) {

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

@ -1383,7 +1383,6 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, SRowKey* pLastPro
static FORCE_INLINE STSchema* getTableSchemaImpl(STsdbReader* pReader, uint64_t uid) {
ASSERT(pReader->info.pSchema == NULL);
int32_t code = metaGetTbTSchemaEx(pReader->pTsdb->pVnode->pMeta, pReader->info.suid, uid, -1, &pReader->info.pSchema);
if (code != TSDB_CODE_SUCCESS || pReader->info.pSchema == NULL) {
terrno = code;
@ -1414,7 +1413,9 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
if (pReader->info.pSchema == NULL) {
pSchema = getTableSchemaImpl(pReader, uid);
if (pSchema == NULL) {
tsdbDebug("%p table uid:%" PRIu64 " has been dropped, no data existed, %s", pReader, uid, pReader->idStr);
code = terrno;
tsdbError("%p table uid:%" PRIu64 " failed to get tableschema, code:%s, %s", pReader, uid, tstrerror(code),
pReader->idStr);
return code;
}
}
@ -1449,7 +1450,7 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
pReader->cost.blockLoadTime += elapsedTime;
pDumpInfo->allDumped = false;
return TSDB_CODE_SUCCESS;
return code;
}
/**
@ -1759,14 +1760,22 @@ static bool tryCopyDistinctRowFromFileBlock(STsdbReader* pReader, SBlockData* pB
return code;
}
static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, int32_t pkSrcSlot,
SVersionRange* pVerRange) {
static int32_t nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, int32_t pkSrcSlot,
SVersionRange* pVerRange) {
int32_t code = 0;
int32_t order = pSttBlockReader->order;
int32_t step = ASCENDING_TRAVERSE(order) ? 1 : -1;
SRowKey* pNextProc = &pScanInfo->sttKeyInfo.nextProcKey;
while (1) {
bool hasVal = tMergeTreeNext(&pSttBlockReader->mergeTree);
bool hasVal = false;
code = tMergeTreeNext(&pSttBlockReader->mergeTree, &hasVal);
if (code) {
tsdbError("failed to iter the next row in stt-file merge tree, code:%s, %s", tstrerror(code),
pSttBlockReader->mergeTree.idStr);
return code;
}
if (!hasVal) { // the next value will be the accessed key in stt
pScanInfo->sttKeyInfo.status = STT_FILE_NO_DATA;
@ -1779,7 +1788,7 @@ static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockSc
memset(pNextProc->pks[0].pData, 0, pNextProc->pks[0].nData);
}
}
return false;
return code;
}
TSDBROW* pRow = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
@ -1798,13 +1807,15 @@ static bool nextRowFromSttBlocks(SSttBlockReader* pSttBlockReader, STableBlockSc
if (!hasBeenDropped(pScanInfo->delSkyline, &pScanInfo->sttBlockDelIndex, key, ver, order, pVerRange,
pSttBlockReader->numOfPks > 0)) {
pScanInfo->sttKeyInfo.status = STT_FILE_HAS_DATA;
return true;
return code;
}
} else {
pScanInfo->sttKeyInfo.status = STT_FILE_HAS_DATA;
return true;
return code;
}
}
return code;
}
static void doPinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreePinSttBlock(&pSttBlockReader->mergeTree); }
@ -1819,9 +1830,14 @@ static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SSttBlockReader* pSttB
// avoid the fetch next row replace the referenced stt block in buffer
doPinSttBlock(pSttBlockReader);
bool hasVal = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
doUnpinSttBlock(pSttBlockReader);
if (hasVal) {
if (code) {
return code;
}
if (hasDataInSttBlock(pScanInfo)) {
SRowKey* pNext = getCurrentKeyInSttBlock(pSttBlockReader);
if (pkCompEx(pSttKey, pNext) != 0) {
code = doAppendRowFromFileBlock(pReader->resBlockInfo.pResBlock, pReader, fRow->pBlockData, fRow->iRow);
@ -2125,7 +2141,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
if (piRow->type == TSDBROW_ROW_FMT) {
piSchema = doGetSchemaForTSRow(TSDBROW_SVERSION(piRow), pReader, pBlockScanInfo->uid);
if (piSchema == NULL) {
return code;
return terrno;
}
}
@ -2380,14 +2396,13 @@ static bool isValidFileBlockRow(SBlockData* pBlockData, int32_t rowIndex, STable
return true;
}
static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
bool hasData = true;
static void initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScanInfo* pScanInfo, STsdbReader* pReader) {
int32_t order = pReader->info.order;
bool asc = ASCENDING_TRAVERSE(order);
// the stt block reader has been initialized for this table.
if (pSttBlockReader->uid == pScanInfo->uid) {
return hasDataInSttBlock(pScanInfo);
return;
}
if (pSttBlockReader->uid != 0) {
@ -2396,9 +2411,14 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
pSttBlockReader->uid = pScanInfo->uid;
// second time init stt block reader
// second or third time init stt block reader
if (pScanInfo->cleanSttBlocks && (pReader->info.execMode == READER_EXEC_ROWS)) {
return !pScanInfo->sttBlockReturned;
// only allowed to retrieve clean stt blocks for count once
if (pScanInfo->sttBlockReturned) {
pScanInfo->sttKeyInfo.status = STT_FILE_NO_DATA;
tsdbDebug("uid:%" PRIu64 " set no stt-file data after stt-block retrieved, %s", pScanInfo->uid, pReader->idStr);
}
return;
}
STimeWindow w = pSttBlockReader->window;
@ -2435,28 +2455,28 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
SSttDataInfoForTable info = {.pKeyRangeList = taosArrayInit(4, sizeof(SSttKeyRange))};
if (info.pKeyRangeList == NULL) {
pReader->code = terrno;
return false;
return;
}
int32_t code = tMergeTreeOpen2(&pSttBlockReader->mergeTree, &conf, &info);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(info.pKeyRangeList);
pReader->code = code;
return false;
return;
}
code = initMemDataIterator(pScanInfo, pReader);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(info.pKeyRangeList);
pReader->code = code;
return false;
return;
}
code = initDelSkylineIterator(pScanInfo, pReader->info.order, &pReader->cost);
if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(info.pKeyRangeList);
pReader->code = code;
return false;
return;
}
if (conf.rspRows) {
@ -2484,27 +2504,26 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
SRowKey* p = asc ? &pScanInfo->sttRange.skey : &pScanInfo->sttRange.ekey;
tRowKeyAssign(&pScanInfo->sttKeyInfo.nextProcKey, p);
hasData = (pScanInfo->sttKeyInfo.status == STT_FILE_HAS_DATA);
} else { // not clean stt blocks
INIT_KEYRANGE(&pScanInfo->sttRange); // reset the time window
pScanInfo->sttBlockReturned = false;
hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
}
} else {
pScanInfo->cleanSttBlocks = false;
INIT_KEYRANGE(&pScanInfo->sttRange); // reset the time window
pScanInfo->sttBlockReturned = false;
hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pReader->suppInfo.pkSrcSlot, &pReader->info.verRange);
}
pScanInfo->sttBlockReturned = false;
taosArrayDestroy(info.pKeyRangeList);
int64_t el = taosGetTimestampUs() - st;
pReader->cost.initSttBlockReader += (el / 1000.0);
tsdbDebug("init stt block reader completed, elapsed time:%" PRId64 "us %s", el, pReader->idStr);
return hasData;
if (code != 0) {
pReader->code = code;
}
}
static bool hasDataInSttBlock(STableBlockScanInfo* pInfo) { return pInfo->sttKeyInfo.status == STT_FILE_HAS_DATA; }
@ -2772,7 +2791,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
}
SBlockData* pBlockData = &pReader->status.fileBlockData;
(void) initSttBlockReader(pSttBlockReader, pBlockScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pBlockScanInfo, pReader);
if (pReader->code != 0) {
code = pReader->code;
goto _end;
@ -3190,12 +3209,12 @@ static int32_t doLoadSttBlockSequentially(STsdbReader* pReader) {
continue;
}
bool hasDataInSttFile = initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
if (pReader->code != TSDB_CODE_SUCCESS) {
return pReader->code;
}
if (!hasDataInSttFile) {
if (!hasDataInSttBlock(pScanInfo)) {
bool hasNexTable = moveToNextTable(pUidList, pStatus);
if (!hasNexTable) {
return TSDB_CODE_SUCCESS;
@ -3287,7 +3306,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
}
if (pScanInfo->sttKeyInfo.status == STT_FILE_READER_UNINIT) {
(void) initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
if (pReader->code != 0) {
return pReader->code;
}
@ -3331,7 +3350,7 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
int64_t st = taosGetTimestampUs();
// let's load data from stt files, make sure clear the cleanStt block flag before load the data from stt files
(void) initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
initSttBlockReader(pSttBlockReader, pScanInfo, pReader);
if (pReader->code != 0) {
return pReader->code;
}
@ -4087,7 +4106,12 @@ int32_t doMergeRowsInSttBlock(SSttBlockReader* pSttBlockReader, STableBlockScanI
SRowKey* pRowKey = &pScanInfo->lastProcKey;
int32_t code = TSDB_CODE_SUCCESS;
while (nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pkSrcSlot, pVerRange)) {
while (1) {
code = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, pkSrcSlot, pVerRange);
if (code || (!hasDataInSttBlock(pScanInfo))) {
return code;
}
SRowKey* pNextKey = getCurrentKeyInSttBlock(pSttBlockReader);
int32_t ret = pkCompEx(pRowKey, pNextKey);

4
source/libs/executor/src/executor.c
View File

@ -663,7 +663,7 @@ int32_t qExecTaskOpt(qTaskInfo_t tinfo, SArray* pResList, uint64_t* useconds, bo
if (isTaskKilled(pTaskInfo)) {
atomic_store_64(&pTaskInfo->owner, 0);
qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));
return TSDB_CODE_SUCCESS;
return pTaskInfo->code;
}
// error occurs, record the error code and return to client
@ -785,7 +785,7 @@ int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds) {
qDebug("%s already killed, abort", GET_TASKID(pTaskInfo));
taosRUnLockLatch(&pTaskInfo->lock);
return TSDB_CODE_SUCCESS;
return pTaskInfo->code;
}
if (pTaskInfo->owner != 0) {

7
source/libs/executor/src/scanoperator.c
View File

@ -464,15 +464,10 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
SSDataBlock* p = NULL;
code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pTableScanInfo->dataReader, &p, NULL);
if (p == NULL || code != TSDB_CODE_SUCCESS) {
if (p == NULL || code != TSDB_CODE_SUCCESS || p != pBlock) {
return code;
}
if(p != pBlock) {
qError("[loadDataBlock] p != pBlock");
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
code = doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
if (code) {
return code;