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

enh: reuse tsdb cache reader

This commit is contained in:
dapan1121 2023-04-14 14:30:20 +08:00
parent 9f78c22fb5
commit 423ec923ca
3 changed files with 27 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/vnode/inc/vnode.h
View File

@ -193,6 +193,7 @@ void *tsdbGetIdx(SMeta *pMeta);
void *tsdbGetIvtIdx(SMeta *pMeta);
uint64_t getReaderMaxVersion(STsdbReader *pReader);
int32_t tsdbReuseCacherowsReader(void* pReader, void* pTableIdList, int32_t numOfTables);
int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList, int32_t numOfTables, int32_t numOfCols,
uint64_t suid, void **pReader, const char *idstr);
int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, SArray *pTableUids);

12
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
View File

@ -142,6 +142,18 @@ static int32_t setTableSchema(SCacheRowsReader* p, uint64_t suid, const char* id
return TSDB_CODE_SUCCESS;
}
int32_t tsdbReuseCacherowsReader(void* reader, void* pTableIdList, int32_t numOfTables) {
SCacheRowsReader* pReader = (SCacheRowsReader*)reader;
pReader->pTableList = pTableIdList;
pReader->numOfTables = numOfTables;
pReader->lastTs = INT64_MIN;
resetLastBlockLoadInfo(pReader->pLoadInfo);
return TSDB_CODE_SUCCESS;
}
int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList, int32_t numOfTables, int32_t numOfCols,
uint64_t suid, void** pReader, const char* idstr) {
*pReader = NULL;

23
source/libs/executor/src/cachescanoperator.c
View File

@ -224,13 +224,17 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
T_LONG_JMP(pTaskInfo->env, code);
}
code = tsdbCacherowsReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
taosArrayGetSize(pInfo->matchInfo.pList), suid, &pInfo->pLastrowReader,
pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) {
pInfo->currentGroupIndex += 1;
taosArrayClear(pInfo->pUidList);
continue;
if (NULL == pInfo->pLastrowReader) {
code = tsdbCacherowsReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pList, num,
taosArrayGetSize(pInfo->matchInfo.pList), suid, &pInfo->pLastrowReader,
pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) {
pInfo->currentGroupIndex += 1;
taosArrayClear(pInfo->pUidList);
continue;
}
} else {
tsdbReuseCacherowsReader(pInfo->pLastrowReader, pList, num);
}
taosArrayClear(pInfo->pUidList);
@ -263,13 +267,14 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
}
}
pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
//pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
return pInfo->pRes;
} else {
pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
//pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
}
}
pInfo->pLastrowReader = tsdbCacherowsReaderClose(pInfo->pLastrowReader);
setOperatorCompleted(pOperator);
return NULL;
}