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

fix(query): if the return value of addTagPseudoColumnData is not success, not jump out.

This commit is contained in:
Haojun Liao 2024-08-16 15:42:53 +08:00
parent 91dfa25dac
commit 0cbbdf1b88
2 changed files with 32 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
source/libs/executor/inc/executorInt.h
View File

@ -856,8 +856,6 @@ int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* de
extern void doDestroyExchangeOperatorInfo(void* param); extern void doDestroyExchangeOperatorInfo(void* param);
int32_t doFilterImpl(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo,
SColumnInfoData** pResCol);
int32_t doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo); int32_t doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo);
int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock, int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int32_t numOfExpr, SSDataBlock* pBlock,
int32_t rows, SExecTaskInfo* pTask, STableMetaCacheInfo* pCache); int32_t rows, SExecTaskInfo* pTask, STableMetaCacheInfo* pCache);

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

@ -279,21 +279,20 @@ static int32_t doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBloc
return code; return code;
} }
static void doSetTagColumnData(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo, static int32_t doSetTagColumnData(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo,
int32_t rows) { int32_t rows) {
if (pTableScanInfo->pseudoSup.numOfExprs > 0) { int32_t code = 0;
SExprSupp* pSup = &pTableScanInfo->pseudoSup; SExprSupp* pSup = &pTableScanInfo->pseudoSup;
if (pSup->numOfExprs > 0) {
int32_t code = addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock, rows, code = addTagPseudoColumnData(&pTableScanInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pBlock, rows,
pTaskInfo, &pTableScanInfo->metaCache); pTaskInfo, &pTableScanInfo->metaCache);
// ignore the table not exists error, since this table may have been dropped during the scan procedure. // ignore the table not exists error, since this table may have been dropped during the scan procedure.
if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_PAR_TABLE_NOT_EXIST) { if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
T_LONG_JMP(pTaskInfo->env, code); code = 0;
} }
// reset the error code.
terrno = 0;
} }
return code;
} }
bool applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) { bool applyLimitOffset(SLimitInfo* pLimitInfo, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {
@ -373,10 +372,10 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64 ", uid:%" PRIu64, qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64 ", uid:%" PRIu64,
GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows,
pBlockInfo->id.uid); pBlockInfo->id.uid);
doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows); code = doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
pCost->skipBlocks += 1; pCost->skipBlocks += 1;
pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
return TSDB_CODE_SUCCESS; QUERY_CHECK_CODE(code, lino, _end);
} else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) { } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {
pCost->loadBlockStatis += 1; pCost->loadBlockStatis += 1;
loadSMA = true; // mark the operation of load sma; loadSMA = true; // mark the operation of load sma;
@ -391,9 +390,9 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
if (success) { // failed to load the block sma data, data block statistics does not exist, load data block instead if (success) { // failed to load the block sma data, data block statistics does not exist, load data block instead
qDebug("%s data block SMA loaded, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, GET_TASKID(pTaskInfo), qDebug("%s data block SMA loaded, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, GET_TASKID(pTaskInfo),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows); code = doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
return TSDB_CODE_SUCCESS; QUERY_CHECK_CODE(code, lino, _end);
} else { } else {
qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo)); qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo));
*status = FUNC_DATA_REQUIRED_DATA_LOAD; *status = FUNC_DATA_REQUIRED_DATA_LOAD;
@ -473,7 +472,11 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
qError("[loadDataBlock] p != pBlock"); qError("[loadDataBlock] p != pBlock");
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR; return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
} }
doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
code = doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
if (code) {
return code;
}
// restore the previous value // restore the previous value
pCost->totalRows -= pBlock->info.rows; pCost->totalRows -= pBlock->info.rows;
@ -912,7 +915,8 @@ static SSDataBlock* getOneRowResultBlock(SExecTaskInfo* pTaskInfo, STableScanBas
} }
// set tag/tbname // set tag/tbname
doSetTagColumnData(pBase, pBlock, pTaskInfo, 1); terrno = doSetTagColumnData(pBase, pBlock, pTaskInfo, 1);
return pBlock; return pBlock;
} }
@ -1633,7 +1637,9 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pReader, &p, NULL); code = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pReader, &p, NULL);
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);
doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows); code = doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);
QUERY_CHECK_CODE(code, lino, _end);
pBlock->info.id.groupId = tableListGetTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid); pBlock->info.id.groupId = tableListGetTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid);
} }
@ -2762,12 +2768,16 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
// currently only the tbname pseudo column // currently only the tbname pseudo column
if (pInfo->numOfPseudoExpr > 0) { if (pInfo->numOfPseudoExpr > 0) {
int32_t tmpCode = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr,
pInfo->pRes, pBlockInfo->rows, pTaskInfo, &pTableScanInfo->base.metaCache); pInfo->pRes, pBlockInfo->rows, pTaskInfo, &pTableScanInfo->base.metaCache);
// ignore the table not exists error, since this table may have been dropped during the scan procedure. // ignore the table not exists error, since this table may have been dropped during the scan procedure.
if (tmpCode != TSDB_CODE_SUCCESS && tmpCode != TSDB_CODE_PAR_TABLE_NOT_EXIST) { if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
code = 0;
}
if (code) {
blockDataFreeRes((SSDataBlock*)pBlock); blockDataFreeRes((SSDataBlock*)pBlock);
T_LONG_JMP(pTaskInfo->env, code); QUERY_CHECK_CODE(code, lino, _end);
} }
// reset the error code. // reset the error code.