From 8e3f5135a66fee65d3f7aa149a6af10705dc7d2f Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Tue, 12 Jul 2022 17:28:14 +0800 Subject: [PATCH] fix(tsdb): close fd --- source/dnode/vnode/src/tsdb/tsdbRead.c | 83 ++++++++++++------------- source/libs/executor/src/executorMain.c | 3 +- source/libs/executor/src/scanoperator.c | 14 +++-- source/libs/stream/src/stream.c | 2 +- source/libs/stream/src/streamDispatch.c | 2 +- source/libs/stream/src/streamExec.c | 4 +- 6 files changed, 56 insertions(+), 52 deletions(-) diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 5f796bbab9..76917066f7 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -63,15 +63,15 @@ typedef struct SBlockLoadSuppInfo { } SBlockLoadSuppInfo; typedef struct SFilesetIter { - int32_t numOfFiles; // number of total files - int32_t index; // current accessed index in the list - SArray* pFileList; // data file list - int32_t order; + int32_t numOfFiles; // number of total files + int32_t index; // current accessed index in the list + SArray* pFileList; // data file list + int32_t order; } SFilesetIter; typedef struct SFileDataBlockInfo { int32_t - tbBlockIdx; // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it + tbBlockIdx; // index position in STableBlockScanInfo in order to check whether neighbor block overlaps with it uint64_t uid; } SFileDataBlockInfo; @@ -119,10 +119,10 @@ struct STsdbReader { int32_t type; // query type: 1. retrieve all data blocks, 2. retrieve direct prev|next rows SBlockLoadSuppInfo suppInfo; - SIOCostSummary cost; - STSchema* pSchema; - SDataFReader* pFileReader; - SVersionRange verRange; + SIOCostSummary cost; + STSchema* pSchema; + SDataFReader* pFileReader; + SVersionRange verRange; }; static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter); @@ -287,9 +287,7 @@ static int32_t initFilesetIterator(SFilesetIter* pIter, const STsdbFSState* pFSt return TSDB_CODE_SUCCESS; } -static void cleanupFilesetIterator(SFilesetIter* pIter) { - taosArrayDestroy(pIter->pFileList); -} +static void cleanupFilesetIterator(SFilesetIter* pIter) { taosArrayDestroy(pIter->pFileList); } static bool filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader) { bool asc = ASCENDING_TRAVERSE(pIter->order); @@ -304,6 +302,7 @@ static bool filesetIteratorNext(SFilesetIter* pIter, STsdbReader* pReader) { STimeWindow win = {0}; while (1) { + /*if (pReader->pFileReader != NULL) tsdbDataFReaderClose(&pReader->pFileReader);*/ pReader->status.pCurrentFileset = (SDFileSet*)taosArrayGet(pIter->pFileList, pIter->index); int32_t code = tsdbDataFReaderOpen(&pReader->pFileReader, pReader->pTsdb, pReader->status.pCurrentFileset); @@ -349,9 +348,7 @@ static void resetDataBlockIterator(SDataBlockIter* pIter, int32_t order) { } } -static void cleanupDataBlockIterator(SDataBlockIter* pIter) { - taosArrayDestroy(pIter->blockList); -} +static void cleanupDataBlockIterator(SDataBlockIter* pIter) { taosArrayDestroy(pIter->blockList); } static void initReaderStatus(SReaderStatus* pStatus) { pStatus->pTableIter = NULL; @@ -392,8 +389,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd initReaderStatus(&pReader->status); - pReader->pTsdb = - getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level); + pReader->pTsdb = getTsdbByRetentions(pVnode, pCond->twindows.skey, pVnode->config.tsdbCfg.retentions, idstr, &level); pReader->suid = pCond->suid; pReader->order = pCond->order; pReader->capacity = 4096; @@ -833,7 +829,7 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI uint8_t *pb = NULL, *pb1 = NULL; int32_t code = tsdbReadColData(pReader->pFileReader, &pBlockScanInfo->blockIdx, pBlock, pSupInfo->colIds, numOfCols, - pBlockData, &pb, &pb1); + pBlockData, &pb, &pb1); if (code != TSDB_CODE_SUCCESS) { goto _error; } @@ -1459,18 +1455,18 @@ static bool overlapWithDelSkyline(STableBlockScanInfo* pBlockScanInfo, const SBl } TSDBKEY* pFirst = taosArrayGet(pBlockScanInfo->delSkyline, 0); - TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline); + TSDBKEY* pLast = taosArrayGetLast(pBlockScanInfo->delSkyline); // ts is not overlap if (pBlock->minKey.ts > pLast->ts || pBlock->maxKey.ts < pFirst->ts) { return false; } - int32_t step = ASCENDING_TRAVERSE(order)? 1:-1; + int32_t step = ASCENDING_TRAVERSE(order) ? 1 : -1; // version is not overlap size_t num = taosArrayGetSize(pBlockScanInfo->delSkyline); - for(int32_t i = pBlockScanInfo->fileDelIndex; i < num; i += step) { + for (int32_t i = pBlockScanInfo->fileDelIndex; i < num; i += step) { TSDBKEY* p = taosArrayGet(pBlockScanInfo->delSkyline, i); if (p->ts >= pBlock->minKey.ts && p->ts <= pBlock->maxKey.ts) { if (p->version >= pBlock->minVersion) { @@ -1502,8 +1498,8 @@ static bool fileBlockShouldLoad(STsdbReader* pReader, SFileDataBlockInfo* pFBloc } // has duplicated ts of different version in this block - bool hasDup = (pBlock->nSubBlock == 1)? pBlock->hasDup:true; - bool overlapWithDel= overlapWithDelSkyline(pScanInfo, pBlock, pReader->order); + bool hasDup = (pBlock->nSubBlock == 1) ? pBlock->hasDup : true; + bool overlapWithDel = overlapWithDelSkyline(pScanInfo, pBlock, pReader->order); return (overlapWithNeighbor || hasDup || dataBlockPartiallyRequired(&pReader->window, &pReader->verRange, pBlock) || keyOverlapFileBlock(key, pBlock, &pReader->verRange) || (pBlock->nRow > pReader->capacity) || overlapWithDel); @@ -2220,17 +2216,18 @@ static STsdb* getTsdbByRetentions(SVnode* pVnode, TSKEY winSKey, SRetention* ret } SVersionRange getQueryVerRange(SVnode* pVnode, SQueryTableDataCond* pCond, int8_t level) { - int64_t startVer = (pCond->startVersion == -1)? 0:pCond->startVersion; + int64_t startVer = (pCond->startVersion == -1) ? 0 : pCond->startVersion; if (VND_IS_RSMA(pVnode)) { return (SVersionRange){.minVer = startVer, .maxVer = tdRSmaGetMaxSubmitVer(pVnode->pSma, level)}; } int64_t endVer = 0; - if (pCond->endVersion == -1) { // user not specified end version, set current maximum version of vnode as the endVersion + if (pCond->endVersion == + -1) { // user not specified end version, set current maximum version of vnode as the endVersion endVer = pVnode->state.applied; } else { - endVer = (pCond->endVersion > pVnode->state.applied)? pVnode->state.applied:pCond->endVersion; + endVer = (pCond->endVersion > pVnode->state.applied) ? pVnode->state.applied : pCond->endVersion; } return (SVersionRange){.minVer = startVer, .maxVer = endVer}; @@ -2274,9 +2271,9 @@ bool hasBeenDropped(const SArray* pDelList, int32_t* index, TSDBKEY* pKey, int32 if (pDelList == NULL) { return false; } - size_t num = taosArrayGetSize(pDelList); - bool asc = ASCENDING_TRAVERSE(order); - int32_t step = asc? 1:-1; + size_t num = taosArrayGetSize(pDelList); + bool asc = ASCENDING_TRAVERSE(order); + int32_t step = asc ? 1 : -1; if (asc) { if (*index >= num - 1) { @@ -2823,7 +2820,7 @@ void tsdbReaderClose(STsdbReader* pReader) { taosMemoryFree(pSupInfo->colIds); taosArrayDestroy(pSupInfo->pColAgg); - for(int32_t i = 0; i < blockDataGetNumOfCols(pReader->pResBlock); ++i) { + for (int32_t i = 0; i < blockDataGetNumOfCols(pReader->pResBlock); ++i) { if (pSupInfo->buildBuf[i] != NULL) { taosMemoryFreeClear(pSupInfo->buildBuf[i]); } @@ -2835,7 +2832,7 @@ void tsdbReaderClose(STsdbReader* pReader) { destroyBlockScanInfo(pReader->status.pTableMap); blockDataDestroy(pReader->pResBlock); - + if (pReader->pFileReader != NULL) tsdbDataFReaderClose(&pReader->pFileReader); #if 0 // if (pReader->status.pTableScanInfo != NULL) { // pReader->status.pTableScanInfo = destroyTableCheckInfo(pReader->status.pTableScanInfo); @@ -3011,8 +3008,8 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) { return TSDB_CODE_SUCCESS; } - pReader->order = pCond->order; - pReader->type = BLOCK_LOAD_OFFSET_ORDER; + pReader->order = pCond->order; + pReader->type = BLOCK_LOAD_OFFSET_ORDER; pReader->status.loadFromFile = true; pReader->status.pTableIter = NULL; @@ -3028,6 +3025,8 @@ int32_t tsdbReaderReset(STsdbReader* pReader, SQueryTableDataCond* pCond) { int32_t numOfTables = 1; SDataBlockIter* pBlockIter = &pReader->status.blockIter; + tsdbDataFReaderClose(&pReader->pFileReader); + STsdbFSState* pFState = pReader->pTsdb->fs->cState; initFilesetIterator(&pReader->status.fileIter, pFState, pReader->order, pReader->idStr); resetDataBlockIterator(&pReader->status.blockIter, pReader->order); @@ -3114,13 +3113,12 @@ int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTa pTableBlockInfo->numOfBlocks += pBlockIter->numOfBlocks; } -/* - hasNext = blockIteratorNext(&pStatus->blockIter); -*/ + /* + hasNext = blockIteratorNext(&pStatus->blockIter); + */ - -// tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pReader, numOfBlocks, numOfTables, -// pReader->pFileGroup->fid, pReader->idStr); + // tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pReader, numOfBlocks, numOfTables, + // pReader->pFileGroup->fid, pReader->idStr); } return code; @@ -3158,7 +3156,7 @@ int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) { return rows; } -int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int64_t *suid) { +int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid) { int32_t sversion = 1; SMetaReader mr = {0}; @@ -3171,7 +3169,7 @@ int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int6 } *suid = 0; - + if (mr.me.type == TSDB_CHILD_TABLE) { *suid = mr.me.ctbEntry.suid; code = metaGetTableEntryByUid(&mr, *suid); @@ -3188,8 +3186,7 @@ int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int6 metaReaderClear(&mr); *pSchema = metaGetTbTSchema(pVnode->pMeta, uid, sversion); - + return TSDB_CODE_SUCCESS; } - diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c index 351846c560..5d9fea523c 100644 --- a/source/libs/executor/src/executorMain.c +++ b/source/libs/executor/src/executorMain.c @@ -288,7 +288,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) { if (!tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus)) { while (1) { uint8_t type = pOperator->operatorType; - /*pOperator->status = OP_OPENED;*/ + pOperator->status = OP_OPENED; if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { SStreamScanInfo* pInfo = pOperator->info; if (pOffset->type == TMQ_OFFSET__LOG) { @@ -326,6 +326,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) { if (pTableInfo->uid == uid) { found = true; pTableScanInfo->currentTable = i; + break; } } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index d70fc548cb..a1831bdff7 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -1221,6 +1221,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } // TODO clean data block if (pInfo->pRes->info.rows > 0) { + qDebug("stream scan log return %d rows", pInfo->pRes->info.rows); return pInfo->pRes; } } else if (ret.fetchType == FETCH_TYPE__META) { @@ -1231,7 +1232,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } else if (ret.fetchType == FETCH_TYPE__NONE) { pTaskInfo->streamInfo.lastStatus = ret.offset; ASSERT(pTaskInfo->streamInfo.lastStatus.version + 1 >= pTaskInfo->streamInfo.prepareStatus.version); - qDebug("stream scan return null"); + qDebug("stream scan log return null"); return NULL; } else { ASSERT(0); @@ -1239,7 +1240,12 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_DATA) { SSDataBlock* pResult = doTableScan(pInfo->pTableScanOp); - return pResult && pResult->info.rows > 0 ? pResult : NULL; + if (pResult && pResult->info.rows > 0) { + qDebug("stream scan tsdb return %d rows", pResult->info.rows); + return pResult; + } + qDebug("stream scan tsdb return null"); + return NULL; } else if (pTaskInfo->streamInfo.prepareStatus.type == TMQ_OFFSET__SNAPSHOT_META) { // TODO scan meta ASSERT(0); @@ -1292,7 +1298,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } return pBlock; } else if (pInfo->blockType == STREAM_INPUT__DATA_SUBMIT) { - qInfo("scan mode %d", pInfo->scanMode); + qDebug("scan mode %d", pInfo->scanMode); if (pInfo->scanMode == STREAM_SCAN_FROM_RES) { blockDataDestroy(pInfo->pUpdateRes); pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE; @@ -1387,7 +1393,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { } } } - qInfo("scan rows: %d", pBlockInfo->rows); + qDebug("scan rows: %d", pBlockInfo->rows); return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes; #if 0 diff --git a/source/libs/stream/src/stream.c b/source/libs/stream/src/stream.c index ed85ce31c3..becfac0cac 100644 --- a/source/libs/stream/src/stream.c +++ b/source/libs/stream/src/stream.c @@ -209,7 +209,7 @@ int32_t streamProcessDispatchReq(SStreamTask* pTask, SStreamDispatchReq* pReq, S int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp) { ASSERT(pRsp->inputStatus == TASK_OUTPUT_STATUS__NORMAL || pRsp->inputStatus == TASK_OUTPUT_STATUS__BLOCKED); - qInfo("task %d receive dispatch rsp", pTask->taskId); + qDebug("task %d receive dispatch rsp", pTask->taskId); int8_t old = atomic_exchange_8(&pTask->outputStatus, pRsp->inputStatus); ASSERT(old == TASK_OUTPUT_STATUS__WAIT); diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 8034840fce..98b0874b00 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -303,7 +303,7 @@ int32_t streamDispatch(SStreamTask* pTask, SMsgCb* pMsgCb) { } ASSERT(pBlock->type == STREAM_INPUT__DATA_BLOCK); - qInfo("stream continue dispatching: task %d", pTask->taskId); + qDebug("stream continue dispatching: task %d", pTask->taskId); SRpcMsg dispatchMsg = {0}; SEpSet* pEpSet = NULL; diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 1286b4c69e..36885e73c0 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -26,12 +26,12 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes) } else if (pItem->type == STREAM_INPUT__DATA_SUBMIT) { ASSERT(pTask->isDataScan); SStreamDataSubmit* pSubmit = (SStreamDataSubmit*)data; - qInfo("task %d %p set submit input %p %p %d", pTask->taskId, pTask, pSubmit, pSubmit->data, *pSubmit->dataRef); + qDebug("task %d %p set submit input %p %p %d", pTask->taskId, pTask, pSubmit, pSubmit->data, *pSubmit->dataRef); qSetStreamInput(exec, pSubmit->data, STREAM_INPUT__DATA_SUBMIT, false); } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) { SStreamDataBlock* pBlock = (SStreamDataBlock*)data; SArray* blocks = pBlock->blocks; - qInfo("task %d %p set ssdata input", pTask->taskId, pTask); + qDebug("task %d %p set ssdata input", pTask->taskId, pTask); qSetMultiStreamInput(exec, blocks->pData, blocks->size, STREAM_INPUT__DATA_BLOCK, false); } else if (pItem->type == STREAM_INPUT__DROP) { // TODO exec drop