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

fix meta dead lock

This commit is contained in:
yihaoDeng 2024-09-11 16:37:06 +08:00
parent f9c8b6cd7b
commit 9b418bee78
1 changed files with 72 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -101,7 +101,7 @@ static int32_t overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBloc
if (order == TSDB_ORDER_ASC) {
w = getAlignQueryTimeWindow(pInterval, pBlockInfo->window.skey);
if(w.ekey < pBlockInfo->window.skey) {
if (w.ekey < pBlockInfo->window.skey) {
qError("w.ekey:%" PRId64 " < pBlockInfo->window.skey:%" PRId64, w.ekey, pBlockInfo->window.skey);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
@ -117,7 +117,7 @@ static int32_t overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBloc
break;
}
if(w.ekey <= pBlockInfo->window.ekey) {
if (w.ekey <= pBlockInfo->window.ekey) {
qError("w.ekey:%" PRId64 " <= pBlockInfo->window.ekey:%" PRId64, w.ekey, pBlockInfo->window.ekey);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
@ -128,7 +128,7 @@ static int32_t overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBloc
}
} else {
w = getAlignQueryTimeWindow(pInterval, pBlockInfo->window.ekey);
if(w.skey > pBlockInfo->window.ekey) {
if (w.skey > pBlockInfo->window.ekey) {
qError("w.skey:%" PRId64 " > pBlockInfo->window.skey:%" PRId64, w.skey, pBlockInfo->window.ekey);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
@ -144,7 +144,7 @@ static int32_t overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBloc
break;
}
if(w.skey >= pBlockInfo->window.skey){
if (w.skey >= pBlockInfo->window.skey) {
qError("w.skey:%" PRId64 " >= pBlockInfo->window.skey:%" PRId64, w.skey, pBlockInfo->window.skey);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
}
@ -399,7 +399,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
}
}
if(*status != FUNC_DATA_REQUIRED_DATA_LOAD) {
if (*status != FUNC_DATA_REQUIRED_DATA_LOAD) {
pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
qError("%s loadDataBlock invalid status:%d", GET_TASKID(pTaskInfo), *status);
return TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
@ -625,8 +625,7 @@ int32_t addTagPseudoColumnData(SReadHandle* pHandle, const SExprInfo* pExpr, int
// append null value before return to caller, since the caller will ignore this error code and proceed
doSetNullValue(pBlock, pExpr, numOfExpr);
} else {
qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(code),
idStr);
qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", pBlock->info.id.uid, tstrerror(code), idStr);
}
pHandle->api.metaReaderFn.clearReader(&mr);
return code;
@ -2804,8 +2803,8 @@ static int32_t setBlockIntoRes(SStreamScanInfo* pInfo, const SSDataBlock* pBlock
// currently only the tbname pseudo column
if (pInfo->numOfPseudoExpr > 0) {
code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr,
pInfo->pRes, pBlockInfo->rows, pTaskInfo, &pTableScanInfo->base.metaCache);
code = addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, 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.
if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {
code = 0;
@ -3471,7 +3470,8 @@ FETCH_NEXT_BLOCK:
(*ppRes) = pInfo->pDeleteDataRes;
return code;
}
qError("%s===stream=== %s failed at line %d since pInfo->pUpdateRes is empty", GET_TASKID(pTaskInfo), __func__, lino);
qError("%s===stream=== %s failed at line %d since pInfo->pUpdateRes is empty", GET_TASKID(pTaskInfo), __func__,
lino);
blockDataCleanup(pInfo->pUpdateDataRes);
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
} break;
@ -3484,7 +3484,8 @@ FETCH_NEXT_BLOCK:
(*ppRes) = pInfo->pUpdateRes;
return code;
}
qError("%s===stream=== %s failed at line %d since pInfo->pUpdateRes is empty", GET_TASKID(pTaskInfo), __func__, lino);
qError("%s===stream=== %s failed at line %d since pInfo->pUpdateRes is empty", GET_TASKID(pTaskInfo), __func__,
lino);
blockDataCleanup(pInfo->pUpdateDataRes);
pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
} break;
@ -4280,7 +4281,6 @@ static int32_t doTagScanOneTable(SOperatorInfo* pOperator, const SSDataBlock* pR
qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),
GET_TASKID(pTaskInfo));
tDecoderClear(&(*mr).coder);
pAPI->metaReaderFn.clearReader(mr);
goto _end;
}
@ -4289,7 +4289,6 @@ static int32_t doTagScanOneTable(SOperatorInfo* pOperator, const SSDataBlock* pR
if (code != TSDB_CODE_SUCCESS) {
qError("failed to get table meta, uid:0x%" PRIx64 ", code:%s, %s", item->uid, tstrerror(terrno),
GET_TASKID(pTaskInfo));
pAPI->metaReaderFn.clearReader(mr);
goto _end;
}
@ -4697,6 +4696,12 @@ static int32_t doTagScanFromMetaEntryNext(SOperatorInfo* pOperator, SSDataBlock*
while (pInfo->curPos < size && count < pOperator->resultInfo.capacity) {
code = doTagScanOneTable(pOperator, pRes, count, &mr, &pTaskInfo->storageAPI);
if (code == TSDB_CODE_OUT_OF_MEMORY) {
break;
} else {
// ignore other error
}
++count;
if (++pInfo->curPos >= size) {
setOperatorCompleted(pOperator);
@ -4814,7 +4819,7 @@ int32_t createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysiNode* p
nodesRewriteExprPostOrder(&pTagCond, tagScanRewriteTagColumn, (void*)&pInfo->filterCtx);
}
}
//TODO wjm check pInfo->filterCtx.code
// TODO wjm check pInfo->filterCtx.code
__optr_fn_t tagScanNextFn = (pTagScanNode->onlyMetaCtbIdx) ? doTagScanFromCtbIdxNext : doTagScanFromMetaEntryNext;
pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, tagScanNextFn, NULL, destroyTagScanOperatorInfo,
optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
@ -5065,8 +5070,7 @@ static int32_t initSubTablesMergeInfo(STableMergeScanInfo* pInfo) {
}
int32_t bufPageSize = pInfo->bufPageSize;
int32_t inMemSize = (pSubTblsInfo->numSubTables - pSubTblsInfo->numTableBlocksInMem) * bufPageSize;
code =
createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);
code = createDiskbasedBuf(&pSubTblsInfo->pBlocksBuf, pInfo->bufPageSize, inMemSize, "blocksExternalBuf", tsTempDir);
if (code != TSDB_CODE_SUCCESS) {
taosMemoryFree(pSubTblsInfo->aInputs);
taosMemoryFree(pSubTblsInfo);
@ -5223,7 +5227,8 @@ _end:
return code;
}
static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity, SSDataBlock** pResBlock) {
static int32_t getSubTablesSortedBlock(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t capacity,
SSDataBlock** pResBlock) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
STableMergeScanInfo* pInfo = pOperator->info;