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

refactor: do some internal refactor.

This commit is contained in:
Haojun Liao 2023-05-29 10:50:50 +08:00
parent 953a69a829
commit 8f61e2d6ef
8 changed files with 45 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
include/libs/executor/storageapi.h
View File

@ -144,25 +144,26 @@ typedef struct {
// int32_t setForSnapShot(SSnapContext *ctx, int64_t uid); // int32_t setForSnapShot(SSnapContext *ctx, int64_t uid);
// int32_t destroySnapContext(SSnapContext *ctx); // int32_t destroySnapContext(SSnapContext *ctx);
// clang-format off
/*-------------------------------------------------new api format---------------------------------------------------*/ /*-------------------------------------------------new api format---------------------------------------------------*/
typedef struct TsdReader { typedef struct TsdReader {
int32_t (*tsdReaderOpen)(void* pVnode, SQueryTableDataCond* pCond, void* pTableList, int32_t numOfTables, int32_t (*tsdReaderOpen)(void* pVnode, SQueryTableDataCond* pCond, void* pTableList, int32_t numOfTables,
SSDataBlock* pResBlock, void** ppReader, const char* idstr, bool countOnly, SSDataBlock* pResBlock, void** ppReader, const char* idstr, bool countOnly,
SHashObj** pIgnoreTables); SHashObj** pIgnoreTables);
void (*tsdReaderClose)(); void (*tsdReaderClose)();
void (*tsdSetReaderTaskId)(void *pReader, const char *pId); void (*tsdSetReaderTaskId)(void *pReader, const char *pId);
int32_t (*tsdSetQueryTableList)(); int32_t (*tsdSetQueryTableList)();
int32_t (*tsdNextDataBlock)(); int32_t (*tsdNextDataBlock)();
int32_t (*tsdReaderRetrieveBlockSMAInfo)(); int32_t (*tsdReaderRetrieveBlockSMAInfo)();
SSDataBlock *(*tsdReaderRetrieveDataBlock)(); SSDataBlock *(*tsdReaderRetrieveDataBlock)();
void (*tsdReaderReleaseDataBlock)(); void (*tsdReaderReleaseDataBlock)();
int32_t (*tsdReaderResetStatus)(); int32_t (*tsdReaderResetStatus)();
int32_t (*tsdReaderGetDataBlockDistInfo)(); int32_t (*tsdReaderGetDataBlockDistInfo)();
int64_t (*tsdReaderGetNumOfInMemRows)(); int64_t (*tsdReaderGetNumOfInMemRows)();
void (*tsdReaderNotifyClosing)(); void (*tsdReaderNotifyClosing)();
} TsdReader; } TsdReader;
typedef struct SStoreCacheReader { typedef struct SStoreCacheReader {
@ -174,6 +175,8 @@ typedef struct SStoreCacheReader {
int32_t (*reuseReader)(void *pReader, void *pTableIdList, int32_t numOfTables); int32_t (*reuseReader)(void *pReader, void *pTableIdList, int32_t numOfTables);
} SStoreCacheReader; } SStoreCacheReader;
// clang-format on
/*------------------------------------------------------------------------------------------------------------------*/ /*------------------------------------------------------------------------------------------------------------------*/
/* /*
void tqReaderSetColIdList(STqReader *pReader, SArray *pColIdList); void tqReaderSetColIdList(STqReader *pReader, SArray *pColIdList);
@ -240,7 +243,6 @@ int32_t metaGetTableTags(SMeta *pMeta, uint64_t suid, SArray *uidList);
const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal); const void *metaGetTableTagVal(void *tag, int16_t type, STagVal *tagVal);
int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName); int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName);
int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid); int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid);
int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType); int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType);
bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid); bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid);
@ -250,13 +252,12 @@ int32_t metaUidFilterCachePut(SMeta *pMeta, uint64_t suid, const void *pKey, in
int32_t payloadLen, double selectivityRatio); int32_t payloadLen, double selectivityRatio);
tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name); tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name);
int32_t metaGetCachedTbGroup(SMeta* pMeta, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList); int32_t metaGetCachedTbGroup(SMeta* pMeta, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList);
int32_t metaPutTbGroupToCache(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload, int32_t metaPutTbGroupToCache(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload, int32_t payloadLen);
int32_t payloadLen);
*/ */
typedef struct SStoreMeta { typedef struct SStoreMeta {
SMTbCursor *(*openTableMetaCursor)(void *pVnode); // metaOpenTbCursor SMTbCursor *(*openTableMetaCursor)(void *pVnode); // metaOpenTbCursor
void (*closeTableMetaCursor)(SMTbCursor *pTbCur); // metaCloseTbCursor void (*closeTableMetaCursor)(SMTbCursor *pTbCur); // metaCloseTbCursor
int32_t (*cursorNext)(SMTbCursor *pTbCur, ETableType jumpTableType); // metaTbCursorNext int32_t (*cursorNext)(SMTbCursor *pTbCur, ETableType jumpTableType); // metaTbCursorNext
int32_t (*cursorPrev)(SMTbCursor *pTbCur, ETableType jumpTableType); // metaTbCursorPrev int32_t (*cursorPrev)(SMTbCursor *pTbCur, ETableType jumpTableType); // metaTbCursorPrev
@ -269,28 +270,20 @@ typedef struct SStoreMeta {
int32_t (*getTableNameByUid)(void *pVnode, uint64_t uid, char *tbName); int32_t (*getTableNameByUid)(void *pVnode, uint64_t uid, char *tbName);
bool (*isTableExisted)(void *pVnode, tb_uid_t uid); bool (*isTableExisted)(void *pVnode, tb_uid_t uid);
/** int32_t (*metaGetCachedTbGroup)(void* pVnode, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList);
int32_t metaGetCachedTbGroup(SMeta* pMeta, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList); int32_t (*metaPutTbGroupToCache)(void* pVnode, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload, int32_t payloadLen);
int32_t metaPutTbGroupToCache(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload,
int32_t payloadLen);
*/
int32_t (*getCachedTableList)(void* pVnode, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray* pList1,
bool* acquireRes);
int32_t (*putCachedTableList)(void* pVnode, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload,
int32_t payloadLen, double selectivityRatio);
// int32_t (*clearCachedTableList)(void* pVnode, uint64_t suid);
/** int32_t (*getCachedTableList)(void* pVnode, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray* pList1, bool* acquireRes);
* int32_t (*putCachedTableList)(void* pVnode, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload, int32_t payloadLen, double selectivityRatio);
*/
void *(*storeGetIndexInfo)();
void *(*getInvertIndex)(void* pVnode);
int32_t (*getChildTableList)(void *pVnode, int64_t suid, SArray *list); // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]
int32_t (*storeGetTableList)(void* pVnode, int8_t type, SArray* pList); // vnodeGetStbIdList & vnodeGetAllTableList
void *storeGetVersionRange;
void *storeGetLastTimestamp;
int32_t (*getTableSchema)(void *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid); // tsdbGetTableSchema void *(*storeGetIndexInfo)();
void *(*getInvertIndex)(void* pVnode);
int32_t (*getChildTableList)(void *pVnode, int64_t suid, SArray *list); // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]
int32_t (*storeGetTableList)(void* pVnode, int8_t type, SArray* pList); // vnodeGetStbIdList & vnodeGetAllTableList
void *storeGetVersionRange;
void *storeGetLastTimestamp;
int32_t (*getTableSchema)(void *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid); // tsdbGetTableSchema
// db name, vgId, numOfTables, numOfSTables // db name, vgId, numOfTables, numOfSTables
int32_t (*getNumOfChildTables)(void* pVnode, int64_t uid, int64_t* numOfTables); // int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo); int32_t (*getNumOfChildTables)(void* pVnode, int64_t uid, int64_t* numOfTables); // int32_t metaGetStbStats(SMeta *pMeta, int64_t uid, SMetaStbStats *pInfo);

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

@ -120,15 +120,13 @@ int32_t metaGetCachedTableUidList(void *pVnode, tb_uid_t suid, const uint8_t *k
int32_t metaUidFilterCachePut(void *pVnode, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload, int32_t metaUidFilterCachePut(void *pVnode, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload,
int32_t payloadLen, double selectivityRatio); int32_t payloadLen, double selectivityRatio);
tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name); tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name);
int32_t metaTbGroupCacheClear(SMeta *pMeta, uint64_t suid); int32_t metaGetCachedTbGroup(void *pVnode, tb_uid_t suid, const uint8_t *pKey, int32_t keyLen, SArray **pList);
int32_t metaGetCachedTbGroup(SMeta *pMeta, tb_uid_t suid, const uint8_t *pKey, int32_t keyLen, SArray **pList); int32_t metaPutTbGroupToCache(void* pVnode, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload,
int32_t metaPutTbGroupToCache(SMeta *pMeta, uint64_t suid, const void *pKey, int32_t keyLen, void *pPayload,
int32_t payloadLen); int32_t payloadLen);
int64_t metaGetTbNum(SMeta *pMeta); int64_t metaGetTbNum(SMeta *pMeta);
int64_t metaGetNtbNum(SMeta *pMeta);
int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables); int32_t metaGetStbStats(void *pVnode, int64_t uid, int64_t *numOfTables);
// tsdb // tsdb
typedef struct STsdbReader STsdbReader; typedef struct STsdbReader STsdbReader;

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

@ -155,6 +155,7 @@ int metaGetTableEntryByName(SMetaReader* pReader, const char* name);
int metaAlterCache(SMeta* pMeta, int32_t nPage); int metaAlterCache(SMeta* pMeta, int32_t nPage);
int32_t metaUidCacheClear(SMeta* pMeta, uint64_t suid); int32_t metaUidCacheClear(SMeta* pMeta, uint64_t suid);
int32_t metaTbGroupCacheClear(SMeta *pMeta, uint64_t suid);
int metaAddIndexToSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq); int metaAddIndexToSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
int metaDropIndexFromSTable(SMeta* pMeta, int64_t version, SDropIndexReq* pReq); int metaDropIndexFromSTable(SMeta* pMeta, int64_t version, SDropIndexReq* pReq);

6
source/dnode/vnode/src/meta/metaCache.c
View File

@ -704,7 +704,8 @@ int32_t metaUidCacheClear(SMeta* pMeta, uint64_t suid) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t metaGetCachedTbGroup(SMeta* pMeta, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList) { int32_t metaGetCachedTbGroup(void* pVnode, tb_uid_t suid, const uint8_t* pKey, int32_t keyLen, SArray** pList) {
SMeta* pMeta = ((SVnode*)pVnode)->pMeta;
int32_t vgId = TD_VID(pMeta->pVnode); int32_t vgId = TD_VID(pMeta->pVnode);
// generate the composed key for LRU cache // generate the composed key for LRU cache
@ -788,9 +789,10 @@ static void freeTbGroupCachePayload(const void* key, size_t keyLen, void* value)
} }
int32_t metaPutTbGroupToCache(SMeta* pMeta, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload, int32_t metaPutTbGroupToCache(void* pVnode, uint64_t suid, const void* pKey, int32_t keyLen, void* pPayload,
int32_t payloadLen) { int32_t payloadLen) {
int32_t code = 0; int32_t code = 0;
SMeta* pMeta = ((SVnode*)pVnode)->pMeta;
int32_t vgId = TD_VID(pMeta->pVnode); int32_t vgId = TD_VID(pMeta->pVnode);
if (payloadLen > tsTagFilterResCacheSize) { if (payloadLen > tsTagFilterResCacheSize) {

2
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -700,8 +700,6 @@ int64_t metaGetTimeSeriesNum(SMeta *pMeta) {
return pMeta->pVnode->config.vndStats.numOfTimeSeries + pMeta->pVnode->config.vndStats.numOfNTimeSeries; return pMeta->pVnode->config.vndStats.numOfTimeSeries + pMeta->pVnode->config.vndStats.numOfNTimeSeries;
} }
int64_t metaGetNtbNum(SMeta *pMeta) { return pMeta->pVnode->config.vndStats.numOfNTables; }
typedef struct { typedef struct {
SMeta *pMeta; SMeta *pMeta;
TBC *pCur; TBC *pCur;

5
source/dnode/vnode/src/vnd/vnodeInitApi.c
View File

@ -31,10 +31,10 @@ static void initSnapshotFn(SStoreSnapshotFn* pSnapshot);
void initStorageAPI(SStorageAPI* pAPI) { void initStorageAPI(SStorageAPI* pAPI) {
initTsdbReaderAPI(&pAPI->tsdReader); initTsdbReaderAPI(&pAPI->tsdReader);
initMetadataAPI(&pAPI->metaFn); initMetadataAPI(&pAPI->metaFn);
initTqAPI(&pAPI->tqReaderFn);
initStateStoreAPI(&pAPI->stateStore); initStateStoreAPI(&pAPI->stateStore);
initMetaReaderAPI(&pAPI->metaReaderFn); initMetaReaderAPI(&pAPI->metaReaderFn);
initMetaFilterAPI(&pAPI->metaFilter); initMetaFilterAPI(&pAPI->metaFilter);
initTqAPI(&pAPI->tqReaderFn);
initFunctionStateStore(&pAPI->functionStore); initFunctionStateStore(&pAPI->functionStore);
initCacheFn(&pAPI->cacheFn); initCacheFn(&pAPI->cacheFn);
initSnapshotFn(&pAPI->snapshotFn); initSnapshotFn(&pAPI->snapshotFn);
@ -91,6 +91,9 @@ void initMetadataAPI(SStoreMeta* pMeta) {
pMeta->getCachedTableList = metaGetCachedTableUidList; pMeta->getCachedTableList = metaGetCachedTableUidList;
pMeta->putCachedTableList = metaUidFilterCachePut; pMeta->putCachedTableList = metaUidFilterCachePut;
pMeta->metaGetCachedTbGroup = metaGetCachedTbGroup;
pMeta->metaPutTbGroupToCache = metaPutTbGroupToCache;
} }
void initTqAPI(SStoreTqReader* pTq) { void initTqAPI(SStoreTqReader* pTq) {

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

@ -494,8 +494,8 @@ int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInf
listNode->pNodeList = group; listNode->pNodeList = group;
genTbGroupDigest((SNode*)listNode, digest, &context); genTbGroupDigest((SNode*)listNode, digest, &context);
nodesFree(listNode); nodesFree(listNode);
ASSERT(0);
// pAPI->metaFn.getCachedTableList(pVnode, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), &tableList, true); pAPI->metaFn.metaGetCachedTbGroup(pVnode, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), &tableList);
if (tableList) { if (tableList) {
taosArrayDestroy(pTableListInfo->pTableList); taosArrayDestroy(pTableListInfo->pTableList);
pTableListInfo->pTableList = tableList; pTableListInfo->pTableList = tableList;
@ -630,10 +630,9 @@ int32_t getColInfoResultForGroupby(void* pVnode, SNodeList* group, STableListInf
info->groupId = calcGroupId(keyBuf, len); info->groupId = calcGroupId(keyBuf, len);
} }
ASSERT(0);
if (tsTagFilterCache) { if (tsTagFilterCache) {
tableList = taosArrayDup(pTableListInfo->pTableList, NULL); tableList = taosArrayDup(pTableListInfo->pTableList, NULL);
// pAPI->metaFn.putCachedTableList(pVnode, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), tableList, taosArrayGetSize(tableList) * sizeof(STableKeyInfo)); pAPI->metaFn.metaPutTbGroupToCache(pVnode, pTableListInfo->idInfo.suid, context.digest, tListLen(context.digest), tableList, taosArrayGetSize(tableList) * sizeof(STableKeyInfo));
} }
// int64_t st2 = taosGetTimestampUs(); // int64_t st2 = taosGetTimestampUs();

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

@ -3446,7 +3446,7 @@ static void buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* pInfo, S
uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName)); uint64_t groupId = calcGroupId(fullStbName, strlen(fullStbName));
pRes->info.id.groupId = groupId; pRes->info.id.groupId = groupId;
int64_t numOfTables = 0;//metaGetNtbNum(pInfo->readHandle.vnode); int64_t numOfTables = 0;
pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables); pAPI->metaFn.getBasicInfo(pInfo->readHandle.vnode, NULL, NULL, NULL, &numOfTables);
if (numOfTables != 0) { if (numOfTables != 0) {