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-23 18:58:54 +08:00
parent a63473b1cc
commit 85afbf0d35
7 changed files with 76 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -225,22 +225,23 @@ typedef int32_t (*__store_reader_open_fn_t)(void *pVnode, SQueryTableDataCond *p
int32_t numOfTables, SSDataBlock *pResBlock, void **ppReader, int32_t numOfTables, SSDataBlock *pResBlock, void **ppReader,
const char *idstr, bool countOnly, SHashObj **pIgnoreTables); const char *idstr, bool countOnly, SHashObj **pIgnoreTables);
typedef struct SStoreTSDReader { typedef struct TsdReader {
__store_reader_open_fn_t storeReaderOpen; __store_reader_open_fn_t tsdReaderOpen;
void (*storeReaderClose)(); void (*tsdReaderClose)();
void (*setReaderId)(void *pReader, const char *pId); void (*tsdSetReaderTaskId)(void *pReader, const char *pId);
void (*storeReaderSetTableList)(); void (*tsdSetQueryTableList)();
int32_t (*storeReaderNextDataBlock)(); int32_t (*tsdReaderNextDataBlock)();
int32_t (*storeReaderRetrieveBlockSMA)();
SSDataBlock *(*storeReaderRetrieveDataBlock)(); int32_t (*tsdReaderRetrieveBlockSMAInfo)();
void (*storeReaderReleaseDataBlock)(); SSDataBlock *(*tsdReaderRetrieveDataBlock)();
void (*storeReaderResetStatus)(); void (*tsdReaderReleaseDataBlock)();
void (*storeReaderGetDataBlockDistInfo)();
void (*storeReaderGetNumOfInMemRows)(); void (*tsdReaderResetStatus)();
void (*storeReaderNotifyClosing)(); void (*tsdReaderGetDataBlockDistInfo)();
} SStoreTSDReader; void (*tsdReaderGetNumOfInMemRows)();
void (*tsdReaderNotifyClosing)();
} TsdReader;
/** /**
* int32_t tsdbReuseCacherowsReader(void* pReader, void* pTableIdList, int32_t numOfTables); * int32_t tsdbReuseCacherowsReader(void* pReader, void* pTableIdList, int32_t numOfTables);
@ -551,7 +552,7 @@ typedef struct SStateStore {
typedef struct SStorageAPI { typedef struct SStorageAPI {
SStoreMeta metaFn; // todo: refactor SStoreMeta metaFn; // todo: refactor
SStoreTSDReader tsdReader; TsdReader tsdReader;
SStoreMetaReader metaReaderFn; SStoreMetaReader metaReaderFn;
SStoreCacheReader cacheFn; SStoreCacheReader cacheFn;
SStoreSnapshotFn snapshotFn; SStoreSnapshotFn snapshotFn;

4
source/libs/executor/inc/executorInt.h
View File

@ -208,7 +208,7 @@ typedef struct STableScanBase {
SLimitInfo limitInfo; SLimitInfo limitInfo;
// there are more than one table list exists in one task, if only one vnode exists. // there are more than one table list exists in one task, if only one vnode exists.
STableListInfo* pTableListInfo; STableListInfo* pTableListInfo;
SStoreTSDReader readerAPI; TsdReader readerAPI;
} STableScanBase; } STableScanBase;
typedef struct STableScanInfo { typedef struct STableScanInfo {
@ -224,7 +224,7 @@ typedef struct STableScanInfo {
int8_t assignBlockUid; int8_t assignBlockUid;
bool hasGroupByTag; bool hasGroupByTag;
bool countOnly; bool countOnly;
SStoreTSDReader readerAPI; TsdReader readerAPI;
} STableScanInfo; } STableScanInfo;
typedef struct STableMergeScanInfo { typedef struct STableMergeScanInfo {

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

@ -166,7 +166,7 @@ void doSetTaskId(SOperatorInfo* pOperator, SStorageAPI *pAPI) {
if (pStreamScanInfo->pTableScanOp != NULL) { if (pStreamScanInfo->pTableScanOp != NULL) {
STableScanInfo* pScanInfo = pStreamScanInfo->pTableScanOp->info; STableScanInfo* pScanInfo = pStreamScanInfo->pTableScanOp->info;
if (pScanInfo->base.dataReader != NULL) { if (pScanInfo->base.dataReader != NULL) {
pAPI->tsdReader.setReaderId(pScanInfo->base.dataReader, pTaskInfo->id.str); pAPI->tsdReader.tsdSetReaderTaskId(pScanInfo->base.dataReader, pTaskInfo->id.str);
} }
} }
} else { } else {
@ -1087,7 +1087,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
if (pOffset->type == TMQ_OFFSET__LOG) { if (pOffset->type == TMQ_OFFSET__LOG) {
// todo refactor: move away // todo refactor: move away
pTaskInfo->storageAPI.tsdReader.storeReaderClose(pScanBaseInfo->dataReader); pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pScanBaseInfo->dataReader);
pScanBaseInfo->dataReader = NULL; pScanBaseInfo->dataReader = NULL;
ASSERT(0); ASSERT(0);
@ -1148,7 +1148,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
pScanInfo->scanTimes = 0; pScanInfo->scanTimes = 0;
if (pScanBaseInfo->dataReader == NULL) { if (pScanBaseInfo->dataReader == NULL) {
int32_t code = pTaskInfo->storageAPI.tsdReader.storeReaderOpen(pScanBaseInfo->readHandle.vnode, &pScanBaseInfo->cond, &keyInfo, 1, int32_t code = pTaskInfo->storageAPI.tsdReader.tsdReaderOpen(pScanBaseInfo->readHandle.vnode, &pScanBaseInfo->cond, &keyInfo, 1,
pScanInfo->pResBlock, (void**) &pScanBaseInfo->dataReader, id, false, NULL); pScanInfo->pResBlock, (void**) &pScanBaseInfo->dataReader, id, false, NULL);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("prepare read tsdb snapshot failed, uid:%" PRId64 ", code:%s %s", pOffset->uid, tstrerror(code), id); qError("prepare read tsdb snapshot failed, uid:%" PRId64 ", code:%s %s", pOffset->uid, tstrerror(code), id);
@ -1159,8 +1159,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
qDebug("tsdb reader created with offset(snapshot) uid:%" PRId64 " ts:%" PRId64 " table index:%d, total:%d, %s", qDebug("tsdb reader created with offset(snapshot) uid:%" PRId64 " ts:%" PRId64 " table index:%d, total:%d, %s",
uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id); uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id);
} else { } else {
pTaskInfo->storageAPI.tsdReader.storeReaderSetTableList(pScanBaseInfo->dataReader, &keyInfo, 1); pTaskInfo->storageAPI.tsdReader.tsdSetQueryTableList(pScanBaseInfo->dataReader, &keyInfo, 1);
pTaskInfo->storageAPI.tsdReader.storeReaderResetStatus(pScanBaseInfo->dataReader, &pScanBaseInfo->cond); pTaskInfo->storageAPI.tsdReader.tsdReaderResetStatus(pScanBaseInfo->dataReader, &pScanBaseInfo->cond);
qDebug("tsdb reader offset seek snapshot to uid:%" PRId64 " ts %" PRId64 " table index:%d numOfTable:%d, %s", qDebug("tsdb reader offset seek snapshot to uid:%" PRId64 " ts %" PRId64 " table index:%d numOfTable:%d, %s",
uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id); uid, pScanBaseInfo->cond.twindows.skey, pScanInfo->currentTable, numOfTables, id);
} }
@ -1189,7 +1189,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
} }
SMetaTableInfo mtInfo = pTaskInfo->storageAPI.snapshotFn.getTableInfoFromSnapshot(sContext); SMetaTableInfo mtInfo = pTaskInfo->storageAPI.snapshotFn.getTableInfoFromSnapshot(sContext);
pTaskInfo->storageAPI.tsdReader.storeReaderClose(pInfo->dataReader); pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pInfo->dataReader);
pInfo->dataReader = NULL; pInfo->dataReader = NULL;
cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);
@ -1207,7 +1207,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0); STableKeyInfo* pList = tableListGetInfo(pTableListInfo, 0);
int32_t size = tableListGetSize(pTableListInfo); int32_t size = tableListGetSize(pTableListInfo);
pTaskInfo->storageAPI.tsdReader.storeReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, NULL, (void**) &pInfo->dataReader, NULL, pTaskInfo->storageAPI.tsdReader.tsdReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, NULL, (void**) &pInfo->dataReader, NULL,
false, NULL); false, NULL);
cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);
@ -1228,7 +1228,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
id); id);
} else if (pOffset->type == TMQ_OFFSET__LOG) { } else if (pOffset->type == TMQ_OFFSET__LOG) {
SStreamRawScanInfo* pInfo = pOperator->info; SStreamRawScanInfo* pInfo = pOperator->info;
pTaskInfo->storageAPI.tsdReader.storeReaderClose(pInfo->dataReader); pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pInfo->dataReader);
pInfo->dataReader = NULL; pInfo->dataReader = NULL;
qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id); qDebug("tmqsnap qStreamPrepareScan snapshot log, %s", id);
} }

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

@ -1195,7 +1195,7 @@ void qStreamCloseTsdbReader(void* task) {
qDebug("wait for the reader stopping"); qDebug("wait for the reader stopping");
} }
pTaskInfo->storageAPI.tsdReader.storeReaderClose(pTSInfo->base.dataReader); pTaskInfo->storageAPI.tsdReader.tsdReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL; pTSInfo->base.dataReader = NULL;
// restore the status, todo refactor. // restore the status, todo refactor.

4
source/libs/executor/src/operator.c
View File

@ -239,7 +239,7 @@ static ERetType doStopDataReader(SOperatorInfo* pOperator, STraverParam* pParam,
STableScanInfo* pInfo = pOperator->info; STableScanInfo* pInfo = pOperator->info;
if (pInfo->base.dataReader != NULL) { if (pInfo->base.dataReader != NULL) {
pAPI->tsdReader.storeReaderNotifyClosing(pInfo->base.dataReader); pAPI->tsdReader.tsdReaderNotifyClosing(pInfo->base.dataReader);
} }
return OPTR_FN_RET_ABORT; return OPTR_FN_RET_ABORT;
} else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
@ -248,7 +248,7 @@ static ERetType doStopDataReader(SOperatorInfo* pOperator, STraverParam* pParam,
if (pInfo->pTableScanOp != NULL) { if (pInfo->pTableScanOp != NULL) {
STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
if (pTableScanInfo != NULL && pTableScanInfo->base.dataReader != NULL) { if (pTableScanInfo != NULL && pTableScanInfo->base.dataReader != NULL) {
pAPI->tsdReader.storeReaderNotifyClosing(pTableScanInfo->base.dataReader); pAPI->tsdReader.tsdReaderNotifyClosing(pTableScanInfo->base.dataReader);
} }
} }

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

@ -267,7 +267,7 @@ static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock,
SStorageAPI* pAPI = &pTaskInfo->storageAPI; SStorageAPI* pAPI = &pTaskInfo->storageAPI;
bool allColumnsHaveAgg = true; bool allColumnsHaveAgg = true;
int32_t code = pAPI->tsdReader.storeReaderRetrieveBlockSMA(pTableScanInfo->dataReader, pBlock, &allColumnsHaveAgg); int32_t code = pAPI->tsdReader.tsdReaderRetrieveBlockSMAInfo(pTableScanInfo->dataReader, pBlock, &allColumnsHaveAgg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
} }
@ -350,7 +350,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->filterOutBlocks += 1; pCost->filterOutBlocks += 1;
pCost->totalRows += pBlock->info.rows; pCost->totalRows += pBlock->info.rows;
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) { } else if (*status == FUNC_DATA_REQUIRED_NOT_LOAD) {
qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64 ", uid:%" PRIu64, qDebug("%s data block skipped, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64 ", uid:%" PRIu64,
@ -358,7 +358,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
pBlockInfo->id.uid); pBlockInfo->id.uid);
doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows); doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
pCost->skipBlocks += 1; pCost->skipBlocks += 1;
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) { } else if (*status == FUNC_DATA_REQUIRED_SMA_LOAD) {
pCost->loadBlockStatis += 1; pCost->loadBlockStatis += 1;
@ -368,7 +368,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
qDebug("%s data block SMA loaded, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, GET_TASKID(pTaskInfo), qDebug("%s data block SMA loaded, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, GET_TASKID(pTaskInfo),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows); doSetTagColumnData(pTableScanInfo, pBlock, pTaskInfo, pBlock->info.rows);
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else { } else {
qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo)); qDebug("%s failed to load SMA, since not all columns have SMA", GET_TASKID(pTaskInfo));
@ -390,7 +390,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
pCost->filterOutBlocks += 1; pCost->filterOutBlocks += 1;
(*status) = FUNC_DATA_REQUIRED_FILTEROUT; (*status) = FUNC_DATA_REQUIRED_FILTEROUT;
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
} }
@ -405,7 +405,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
qDebug("%s data block skipped due to dynamic prune, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64, qDebug("%s data block skipped due to dynamic prune, brange:%" PRId64 "-%" PRId64 ", rows:%" PRId64,
GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
pCost->skipBlocks += 1; pCost->skipBlocks += 1;
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->dataReader);
STableScanInfo* p1 = pOperator->info; STableScanInfo* p1 = pOperator->info;
if (taosHashGetSize(p1->pIgnoreTables) == taosArrayGetSize(p1->base.pTableListInfo->pTableList)) { if (taosHashGetSize(p1->pIgnoreTables) == taosArrayGetSize(p1->base.pTableListInfo->pTableList)) {
@ -419,7 +419,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca
pCost->totalCheckedRows += pBlock->info.rows; pCost->totalCheckedRows += pBlock->info.rows;
pCost->loadBlocks += 1; pCost->loadBlocks += 1;
SSDataBlock* p = pAPI->tsdReader.storeReaderRetrieveDataBlock(pTableScanInfo->dataReader, NULL); SSDataBlock* p = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pTableScanInfo->dataReader, NULL);
if (p == NULL) { if (p == NULL) {
return terrno; return terrno;
} }
@ -693,9 +693,9 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
while (true) { while (true) {
code = pAPI->tsdReader.storeReaderNextDataBlock(pTableScanInfo->base.dataReader, &hasNext); code = pAPI->tsdReader.tsdReaderNextDataBlock(pTableScanInfo->base.dataReader, &hasNext);
if (code) { if (code) {
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->base.dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
} }
@ -704,12 +704,12 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) {
} }
if (isTaskKilled(pTaskInfo)) { if (isTaskKilled(pTaskInfo)) {
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->base.dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code); T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
} }
if (pOperator->status == OP_EXEC_DONE) { if (pOperator->status == OP_EXEC_DONE) {
pAPI->tsdReader.storeReaderReleaseDataBlock(pTableScanInfo->base.dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pTableScanInfo->base.dataReader);
break; break;
} }
@ -774,7 +774,7 @@ static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) {
qDebug("start to repeat ascending order scan data blocks due to query func required, %s", GET_TASKID(pTaskInfo)); qDebug("start to repeat ascending order scan data blocks due to query func required, %s", GET_TASKID(pTaskInfo));
// do prepare for the next round table scan operation // do prepare for the next round table scan operation
pAPI->tsdReader.storeReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond); pAPI->tsdReader.tsdReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
} }
} }
@ -782,7 +782,7 @@ static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) {
if (pTableScanInfo->scanTimes < total) { if (pTableScanInfo->scanTimes < total) {
if (pTableScanInfo->base.cond.order == TSDB_ORDER_ASC) { if (pTableScanInfo->base.cond.order == TSDB_ORDER_ASC) {
prepareForDescendingScan(&pTableScanInfo->base, pOperator->exprSupp.pCtx, 0); prepareForDescendingScan(&pTableScanInfo->base, pOperator->exprSupp.pCtx, 0);
pAPI->tsdReader.storeReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond); pAPI->tsdReader.tsdReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo)); qDebug("%s start to descending order scan data blocks due to query func required", GET_TASKID(pTaskInfo));
} }
@ -800,7 +800,7 @@ static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) {
pTableScanInfo->base.scanFlag = MAIN_SCAN; pTableScanInfo->base.scanFlag = MAIN_SCAN;
qDebug("%s start to repeat descending order scan data blocks", GET_TASKID(pTaskInfo)); qDebug("%s start to repeat descending order scan data blocks", GET_TASKID(pTaskInfo));
pAPI->tsdReader.storeReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond); pAPI->tsdReader.tsdReaderResetStatus(pTableScanInfo->base.dataReader, &pTableScanInfo->base.cond);
} }
} }
} }
@ -839,11 +839,11 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
tInfo = *(STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->currentTable); tInfo = *(STableKeyInfo*)tableListGetInfo(pInfo->base.pTableListInfo, pInfo->currentTable);
taosRUnLockLatch(&pTaskInfo->lock); taosRUnLockLatch(&pTaskInfo->lock);
pAPI->tsdReader.storeReaderSetTableList(pInfo->base.dataReader, &tInfo, 1); pAPI->tsdReader.tsdSetQueryTableList(pInfo->base.dataReader, &tInfo, 1);
qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d/%d %s", tInfo.uid, numOfTables, qDebug("set uid:%" PRIu64 " into scanner, total tables:%d, index:%d/%d %s", tInfo.uid, numOfTables,
pInfo->currentTable, numOfTables, GET_TASKID(pTaskInfo)); pInfo->currentTable, numOfTables, GET_TASKID(pTaskInfo));
pAPI->tsdReader.storeReaderResetStatus(pInfo->base.dataReader, &pInfo->base.cond); pAPI->tsdReader.tsdReaderResetStatus(pInfo->base.dataReader, &pInfo->base.cond);
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
} }
} else { // scan table group by group sequentially } else { // scan table group by group sequentially
@ -858,7 +858,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
tableListGetGroupList(pInfo->base.pTableListInfo, pInfo->currentGroupId, &pList, &num); tableListGetGroupList(pInfo->base.pTableListInfo, pInfo->currentGroupId, &pList, &num);
ASSERT(pInfo->base.dataReader == NULL); ASSERT(pInfo->base.dataReader == NULL);
int32_t code = pAPI->tsdReader.storeReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock, int32_t code = pAPI->tsdReader.tsdReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock,
(void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), pInfo->countOnly, &pInfo->pIgnoreTables); (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), pInfo->countOnly, &pInfo->pIgnoreTables);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
@ -887,8 +887,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
STableKeyInfo* pList = NULL; STableKeyInfo* pList = NULL;
tableListGetGroupList(pInfo->base.pTableListInfo, pInfo->currentGroupId, &pList, &num); tableListGetGroupList(pInfo->base.pTableListInfo, pInfo->currentGroupId, &pList, &num);
pAPI->tsdReader.storeReaderSetTableList(pInfo->base.dataReader, pList, num); pAPI->tsdReader.tsdSetQueryTableList(pInfo->base.dataReader, pList, num);
pAPI->tsdReader.storeReaderResetStatus(pInfo->base.dataReader, &pInfo->base.cond); pAPI->tsdReader.tsdReaderResetStatus(pInfo->base.dataReader, &pInfo->base.cond);
pInfo->scanTimes = 0; pInfo->scanTimes = 0;
result = doGroupedTableScan(pOperator); result = doGroupedTableScan(pOperator);
@ -910,10 +910,10 @@ static int32_t getTableScannerExecInfo(struct SOperatorInfo* pOptr, void** pOptr
return 0; return 0;
} }
static void destroyTableScanBase(STableScanBase* pBase, SStoreTSDReader* pAPI) { static void destroyTableScanBase(STableScanBase* pBase, TsdReader* pAPI) {
cleanupQueryTableDataCond(&pBase->cond); cleanupQueryTableDataCond(&pBase->cond);
pAPI->storeReaderClose(pBase->dataReader); pAPI->tsdReaderClose(pBase->dataReader);
pBase->dataReader = NULL; pBase->dataReader = NULL;
if (pBase->matchInfo.pList != NULL) { if (pBase->matchInfo.pList != NULL) {
@ -1074,7 +1074,7 @@ void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin, uint6
pTableScanInfo->base.cond.endVersion = version; pTableScanInfo->base.cond.endVersion = version;
pTableScanInfo->scanTimes = 0; pTableScanInfo->scanTimes = 0;
pTableScanInfo->currentGroupId = -1; pTableScanInfo->currentGroupId = -1;
pTableScanInfo->readerAPI.storeReaderClose(pTableScanInfo->base.dataReader); pTableScanInfo->readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);
pTableScanInfo->base.dataReader = NULL; pTableScanInfo->base.dataReader = NULL;
} }
@ -1094,7 +1094,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
SSDataBlock* pBlock = pTableScanInfo->pResBlock; SSDataBlock* pBlock = pTableScanInfo->pResBlock;
STsdbReader* pReader = NULL; STsdbReader* pReader = NULL;
int32_t code = pAPI->tsdReader.storeReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock, int32_t code = pAPI->tsdReader.tsdReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock,
(void**)&pReader, GET_TASKID(pTaskInfo), false, NULL); (void**)&pReader, GET_TASKID(pTaskInfo), false, NULL);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
terrno = code; terrno = code;
@ -1103,7 +1103,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
} }
bool hasNext = false; bool hasNext = false;
code = pAPI->tsdReader.storeReaderNextDataBlock(pReader, &hasNext); code = pAPI->tsdReader.tsdReaderNextDataBlock(pReader, &hasNext);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
terrno = code; terrno = code;
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
@ -1111,12 +1111,12 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU
} }
if (hasNext) { if (hasNext) {
/*SSDataBlock* p = */ pAPI->tsdReader.storeReaderRetrieveDataBlock(pReader, NULL); /*SSDataBlock* p = */ pAPI->tsdReader.tsdReaderRetrieveDataBlock(pReader, NULL);
doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows); doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows);
pBlock->info.id.groupId = getTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid); pBlock->info.id.groupId = getTableGroupId(pTableScanInfo->base.pTableListInfo, pBlock->info.id.uid);
} }
pAPI->tsdReader.storeReaderClose(pReader); pAPI->tsdReader.tsdReaderClose(pReader);
qDebug("retrieve prev rows:%" PRId64 ", skey:%" PRId64 ", ekey:%" PRId64 " uid:%" PRIu64 ", max ver:%" PRId64 qDebug("retrieve prev rows:%" PRId64 ", skey:%" PRId64 ", ekey:%" PRId64 " uid:%" PRIu64 ", max ver:%" PRId64
", suid:%" PRIu64, ", suid:%" PRIu64,
pBlock->info.rows, startTs, endTs, tbUid, maxVersion, cond.suid); pBlock->info.rows, startTs, endTs, tbUid, maxVersion, cond.suid);
@ -1249,7 +1249,7 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
*pRowIndex = 0; *pRowIndex = 0;
pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX}; pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
pTableScanInfo->readerAPI.storeReaderClose(pTableScanInfo->base.dataReader); pTableScanInfo->readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);
pTableScanInfo->base.dataReader = NULL; pTableScanInfo->base.dataReader = NULL;
return NULL; return NULL;
} }
@ -1657,7 +1657,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
return pResult; return pResult;
} }
STableScanInfo* pTSInfo = pInfo->pTableScanOp->info; STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
pAPI->tsdReader.storeReaderClose(pTSInfo->base.dataReader); pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL; pTSInfo->base.dataReader = NULL;
qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1); qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1);
@ -1821,7 +1821,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN2; pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN2;
} }
pAPI->tsdReader.storeReaderClose(pTSInfo->base.dataReader); pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL; pTSInfo->base.dataReader = NULL;
pInfo->pTableScanOp->status = OP_OPENED; pInfo->pTableScanOp->status = OP_OPENED;
@ -1901,7 +1901,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
} }
pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE; pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;
STableScanInfo* pTSInfo = pInfo->pTableScanOp->info; STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
pAPI->tsdReader.storeReaderClose(pTSInfo->base.dataReader); pAPI->tsdReader.tsdReaderClose(pTSInfo->base.dataReader);
pTSInfo->base.dataReader = NULL; pTSInfo->base.dataReader = NULL;
@ -2162,20 +2162,20 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) { if (pTaskInfo->streamInfo.currentOffset.type == TMQ_OFFSET__SNAPSHOT_DATA) {
bool hasNext = false; bool hasNext = false;
if (pInfo->dataReader) { if (pInfo->dataReader) {
code = pAPI->tsdReader.storeReaderNextDataBlock(pInfo->dataReader, &hasNext); code = pAPI->tsdReader.tsdReaderNextDataBlock(pInfo->dataReader, &hasNext);
if (code) { if (code) {
pAPI->tsdReader.storeReaderReleaseDataBlock(pInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->dataReader);
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
} }
} }
if (pInfo->dataReader && hasNext) { if (pInfo->dataReader && hasNext) {
if (isTaskKilled(pTaskInfo)) { if (isTaskKilled(pTaskInfo)) {
pAPI->tsdReader.storeReaderReleaseDataBlock(pInfo->dataReader); pAPI->tsdReader.tsdReaderReleaseDataBlock(pInfo->dataReader);
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code); T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
} }
SSDataBlock* pBlock = pAPI->tsdReader.storeReaderRetrieveDataBlock(pInfo->dataReader, NULL); SSDataBlock* pBlock = pAPI->tsdReader.tsdReaderRetrieveDataBlock(pInfo->dataReader, NULL);
if (pBlock == NULL) { if (pBlock == NULL) {
T_LONG_JMP(pTaskInfo->env, terrno); T_LONG_JMP(pTaskInfo->env, terrno);
} }
@ -2227,7 +2227,7 @@ static SSDataBlock* doRawScan(SOperatorInfo* pOperator) {
static void destroyRawScanOperatorInfo(void* param) { static void destroyRawScanOperatorInfo(void* param) {
SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param; SStreamRawScanInfo* pRawScan = (SStreamRawScanInfo*)param;
pRawScan->pAPI->tsdReader.storeReaderClose(pRawScan->dataReader); pRawScan->pAPI->tsdReader.tsdReaderClose(pRawScan->dataReader);
pRawScan->pAPI->snapshotFn.destroySnapshot(pRawScan->sContext); pRawScan->pAPI->snapshotFn.destroySnapshot(pRawScan->sContext);
tableListDestroy(pRawScan->pTableListInfo); tableListDestroy(pRawScan->pTableListInfo);
taosMemoryFree(pRawScan); taosMemoryFree(pRawScan);
@ -2662,7 +2662,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
SReadHandle* pHandle = &pInfo->base.readHandle; SReadHandle* pHandle = &pInfo->base.readHandle;
if (NULL == source->dataReader || !source->multiReader) { if (NULL == source->dataReader || !source->multiReader) {
code = pAPI->tsdReader.storeReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, (void**)&source->dataReader, GET_TASKID(pTaskInfo), false, NULL); code = pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, (void**)&source->dataReader, GET_TASKID(pTaskInfo), false, NULL);
if (code != 0) { if (code != 0) {
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
} }
@ -2674,9 +2674,9 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
qTrace("tsdb/read-table-data: %p, enter next reader", reader); qTrace("tsdb/read-table-data: %p, enter next reader", reader);
while (true) { while (true) {
code = pAPI->tsdReader.storeReaderNextDataBlock(reader, &hasNext); code = pAPI->tsdReader.tsdReaderNextDataBlock(reader, &hasNext);
if (code != 0) { if (code != 0) {
pAPI->tsdReader.storeReaderReleaseDataBlock(reader); pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);
pInfo->base.dataReader = NULL; pInfo->base.dataReader = NULL;
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
} }
@ -2686,7 +2686,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
} }
if (isTaskKilled(pTaskInfo)) { if (isTaskKilled(pTaskInfo)) {
pAPI->tsdReader.storeReaderReleaseDataBlock(reader); pAPI->tsdReader.tsdReaderReleaseDataBlock(reader);
pInfo->base.dataReader = NULL; pInfo->base.dataReader = NULL;
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code); T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
} }
@ -2726,7 +2726,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
qTrace("tsdb/read-table-data: %p, close reader", reader); qTrace("tsdb/read-table-data: %p, close reader", reader);
if (!source->multiReader) { if (!source->multiReader) {
pAPI->tsdReader.storeReaderClose(pInfo->base.dataReader); pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);
source->dataReader = NULL; source->dataReader = NULL;
} }
pInfo->base.dataReader = NULL; pInfo->base.dataReader = NULL;
@ -2734,7 +2734,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
} }
if (!source->multiReader) { if (!source->multiReader) {
pAPI->tsdReader.storeReaderClose(pInfo->base.dataReader); pAPI->tsdReader.tsdReaderClose(pInfo->base.dataReader);
source->dataReader = NULL; source->dataReader = NULL;
} }
pInfo->base.dataReader = NULL; pInfo->base.dataReader = NULL;
@ -2853,7 +2853,7 @@ int32_t stopGroupTableMergeScan(SOperatorInfo* pOperator) {
for (int32_t i = 0; i < numOfTable; ++i) { for (int32_t i = 0; i < numOfTable; ++i) {
STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i); STableMergeScanSortSourceParam* param = taosArrayGet(pInfo->sortSourceParams, i);
blockDataDestroy(param->inputBlock); blockDataDestroy(param->inputBlock);
pAPI->tsdReader.storeReaderClose(param->dataReader); pAPI->tsdReader.tsdReaderClose(param->dataReader);
param->dataReader = NULL; param->dataReader = NULL;
} }
taosArrayClear(pInfo->sortSourceParams); taosArrayClear(pInfo->sortSourceParams);
@ -2965,11 +2965,11 @@ void destroyTableMergeScanOperatorInfo(void* param) {
for (int32_t i = 0; i < numOfTable; i++) { for (int32_t i = 0; i < numOfTable; i++) {
STableMergeScanSortSourceParam* p = taosArrayGet(pTableScanInfo->sortSourceParams, i); STableMergeScanSortSourceParam* p = taosArrayGet(pTableScanInfo->sortSourceParams, i);
blockDataDestroy(p->inputBlock); blockDataDestroy(p->inputBlock);
pTableScanInfo->base.readerAPI.storeReaderClose(p->dataReader); pTableScanInfo->base.readerAPI.tsdReaderClose(p->dataReader);
p->dataReader = NULL; p->dataReader = NULL;
} }
pTableScanInfo->base.readerAPI.storeReaderClose(pTableScanInfo->base.dataReader); pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);
pTableScanInfo->base.dataReader = NULL; pTableScanInfo->base.dataReader = NULL;
taosArrayDestroy(pTableScanInfo->sortSourceParams); taosArrayDestroy(pTableScanInfo->sortSourceParams);

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

@ -2198,7 +2198,7 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) {
T_LONG_JMP(pTaskInfo->env, code); T_LONG_JMP(pTaskInfo->env, code);
} }
pAPI->tsdReader.storeReaderGetDataBlockDistInfo(pBlockScanInfo->pHandle, &blockDistInfo); pAPI->tsdReader.tsdReaderGetDataBlockDistInfo(pBlockScanInfo->pHandle, &blockDistInfo);
blockDistInfo.numOfInmemRows = (int32_t) pAPI->metaFn.getNumOfRowsInMem(pBlockScanInfo->pHandle); blockDistInfo.numOfInmemRows = (int32_t) pAPI->metaFn.getNumOfRowsInMem(pBlockScanInfo->pHandle);
SSDataBlock* pBlock = pBlockScanInfo->pResBlock; SSDataBlock* pBlock = pBlockScanInfo->pResBlock;
@ -2229,7 +2229,7 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo* pOperator) {
static void destroyBlockDistScanOperatorInfo(void* param) { static void destroyBlockDistScanOperatorInfo(void* param) {
SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param; SBlockDistInfo* pDistInfo = (SBlockDistInfo*)param;
blockDataDestroy(pDistInfo->pResBlock); blockDataDestroy(pDistInfo->pResBlock);
pDistInfo->readHandle.api.tsdReader.storeReaderClose(pDistInfo->pHandle); pDistInfo->readHandle.api.tsdReader.tsdReaderClose(pDistInfo->pHandle);
tableListDestroy(pDistInfo->pTableListInfo); tableListDestroy(pDistInfo->pTableListInfo);
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }
@ -2284,7 +2284,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
size_t num = tableListGetSize(pTableListInfo); size_t num = tableListGetSize(pTableListInfo);
void* pList = tableListGetInfo(pTableListInfo, 0); void* pList = tableListGetInfo(pTableListInfo, 0);
code = readHandle->api.tsdReader.storeReaderOpen(readHandle->vnode, &cond, pList, num, pInfo->pResBlock, (void**)&pInfo->pHandle, pTaskInfo->id.str, false, NULL); code = readHandle->api.tsdReader.tsdReaderOpen(readHandle->vnode, &cond, pList, num, pInfo->pResBlock, (void**)&pInfo->pHandle, pTaskInfo->id.str, false, NULL);
cleanupQueryTableDataCond(&cond); cleanupQueryTableDataCond(&cond);
if (code != 0) { if (code != 0) {
goto _error; goto _error;