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

fix: sys table scan and meta query

This commit is contained in:
kailixu 2024-03-22 10:33:46 +08:00
parent 61b680241d
commit fb2667d92d
4 changed files with 17 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -238,12 +238,12 @@ typedef struct SStoreSnapshotFn {
} SStoreSnapshotFn;
typedef struct SStoreMeta {
SMTbCursor* (*openTableMetaCursor)(void* pVnode); // metaOpenTbCursor
void (*closeTableMetaCursor)(SMTbCursor* pTbCur); // metaCloseTbCursor
void (*pauseTableMetaCursor)(SMTbCursor* pTbCur); // metaPauseTbCursor
void (*resumeTableMetaCursor)(SMTbCursor* pTbCur, int8_t first); // metaResumeTbCursor
int32_t (*cursorNext)(SMTbCursor* pTbCur, ETableType jumpTableType); // metaTbCursorNext
int32_t (*cursorPrev)(SMTbCursor* pTbCur, ETableType jumpTableType); // metaTbCursorPrev
SMTbCursor* (*openTableMetaCursor)(void* pVnode); // metaOpenTbCursor
void (*closeTableMetaCursor)(SMTbCursor* pTbCur); // metaCloseTbCursor
void (*pauseTableMetaCursor)(SMTbCursor* pTbCur); // metaPauseTbCursor
void (*resumeTableMetaCursor)(SMTbCursor* pTbCur, int8_t first, int8_t move); // metaResumeTbCursor
int32_t (*cursorNext)(SMTbCursor* pTbCur, ETableType jumpTableType); // metaTbCursorNext
int32_t (*cursorPrev)(SMTbCursor* pTbCur, ETableType jumpTableType); // metaTbCursorPrev
int32_t (*getTableTags)(void* pVnode, uint64_t suid, SArray* uidList);
int32_t (*getTableTagsByUid)(void* pVnode, int64_t suid, SArray* uidList);

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

@ -136,7 +136,7 @@ typedef SVCreateTSmaReq SSmaCfg;
SMTbCursor* metaOpenTbCursor(void* pVnode);
void metaCloseTbCursor(SMTbCursor* pTbCur);
void metaPauseTbCursor(SMTbCursor* pTbCur);
void metaResumeTbCursor(SMTbCursor* pTbCur, int8_t first);
void metaResumeTbCursor(SMTbCursor* pTbCur, int8_t first, int8_t move);
int32_t metaTbCursorNext(SMTbCursor* pTbCur, ETableType jumpTableType);
int32_t metaTbCursorPrev(SMTbCursor* pTbCur, ETableType jumpTableType);

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

@ -248,7 +248,7 @@ SMTbCursor *metaOpenTbCursor(void *pVnode) {
// tdbTbcMoveToFirst((TBC *)pTbCur->pDbc);
pTbCur->pMeta = pVnodeObj->pMeta;
pTbCur->paused = 1;
metaResumeTbCursor(pTbCur, 1);
metaResumeTbCursor(pTbCur, 1, 0);
return pTbCur;
}
@ -273,7 +273,7 @@ void metaPauseTbCursor(SMTbCursor *pTbCur) {
pTbCur->paused = 1;
}
}
void metaResumeTbCursor(SMTbCursor *pTbCur, int8_t first) {
void metaResumeTbCursor(SMTbCursor *pTbCur, int8_t first, int8_t move) {
if (pTbCur->paused) {
metaReaderDoInit(&pTbCur->mr, pTbCur->pMeta, META_READER_LOCK);
@ -282,9 +282,11 @@ void metaResumeTbCursor(SMTbCursor *pTbCur, int8_t first) {
if (first) {
tdbTbcMoveToFirst((TBC *)pTbCur->pDbc);
} else {
int c = 0;
int c = 1;
tdbTbcMoveTo(pTbCur->pDbc, pTbCur->pKey, pTbCur->kLen, &c);
if (c < 0) {
if (c == 0) {
if (move) tdbTbcMoveToNext(pTbCur->pDbc);
} else if (c < 0) {
tdbTbcMoveToPrev(pTbCur->pDbc);
} else {
tdbTbcMoveToNext(pTbCur->pDbc);

16
source/libs/executor/src/sysscanoperator.c
View File

@ -539,7 +539,7 @@ static SSDataBlock* sysTableScanUserCols(SOperatorInfo* pOperator) {
if (pInfo->pCur == NULL) {
pInfo->pCur = pAPI->metaFn.openTableMetaCursor(pInfo->readHandle.vnode);
} else {
pAPI->metaFn.resumeTableMetaCursor(pInfo->pCur, 0);
pAPI->metaFn.resumeTableMetaCursor(pInfo->pCur, 0, 0);
}
if (pInfo->pSchema == NULL) {
@ -555,14 +555,8 @@ static SSDataBlock* sysTableScanUserCols(SOperatorInfo* pOperator) {
return NULL;
}
int32_t restore = pInfo->restore;
pInfo->restore = false;
while (restore || ((ret = pAPI->metaFn.cursorNext(pInfo->pCur, TSDB_TABLE_MAX)) == 0)) {
if (restore) {
restore = false;
}
while (((ret = pAPI->metaFn.cursorNext(pInfo->pCur, TSDB_TABLE_MAX)) == 0)) {
char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
char tableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
@ -616,7 +610,6 @@ static SSDataBlock* sysTableScanUserCols(SOperatorInfo* pOperator) {
if ((numOfRows + schemaRow->nCols) > pOperator->resultInfo.capacity) {
relocateAndFilterSysTagsScanResult(pInfo, numOfRows, pDataBlock, pOperator->exprSupp.pFilterInfo);
numOfRows = 0;
pInfo->restore = true;
if (pInfo->pRes->info.rows > 0) {
pAPI->metaFn.pauseTableMetaCursor(pInfo->pCur);
@ -724,7 +717,7 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
if (pInfo->pCur == NULL) {
pInfo->pCur = pAPI->metaFn.openTableMetaCursor(pInfo->readHandle.vnode);
} else {
pAPI->metaFn.resumeTableMetaCursor(pInfo->pCur, 0);
pAPI->metaFn.resumeTableMetaCursor(pInfo->pCur, 0, 0);
}
bool blockFull = false;
@ -750,7 +743,6 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
}
if ((smrSuperTable.me.stbEntry.schemaTag.nCols + numOfRows) > pOperator->resultInfo.capacity) {
pAPI->metaFn.cursorPrev(pInfo->pCur, TSDB_TABLE_MAX);
blockFull = true;
} else {
sysTableUserTagsFillOneTableTags(pInfo, &smrSuperTable, &pInfo->pCur->mr, dbname, tableName, &numOfRows,
@ -1354,7 +1346,7 @@ static SSDataBlock* sysTableBuildUserTables(SOperatorInfo* pOperator) {
firstMetaCursor = 1;
}
if (!firstMetaCursor) {
pAPI->metaFn.resumeTableMetaCursor(pInfo->pCur, 0);
pAPI->metaFn.resumeTableMetaCursor(pInfo->pCur, 0, 1);
}
blockDataCleanup(pInfo->pRes);