From 0dd0b16842be80da91c90909d24df680ee4f623c Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Fri, 12 May 2023 13:30:17 +0800 Subject: [PATCH] enh: optimize table scan for last query --- include/libs/function/functionMgt.h | 1 + source/dnode/vnode/inc/vnode.h | 2 +- source/dnode/vnode/src/tsdb/tsdbRead.c | 41 +++++++++++++++++--- source/libs/executor/inc/executil.h | 20 +++++++++- source/libs/executor/inc/executorimpl.h | 1 + source/libs/executor/src/executil.c | 17 -------- source/libs/executor/src/executor.c | 4 +- source/libs/executor/src/scanoperator.c | 45 +++++++++++++++++++--- source/libs/executor/src/sysscanoperator.c | 2 +- 9 files changed, 100 insertions(+), 33 deletions(-) diff --git a/include/libs/function/functionMgt.h b/include/libs/function/functionMgt.h index 42bc89f0b7..f097a2b2df 100644 --- a/include/libs/function/functionMgt.h +++ b/include/libs/function/functionMgt.h @@ -230,6 +230,7 @@ typedef enum EFuncDataRequired { FUNC_DATA_REQUIRED_SMA_LOAD, FUNC_DATA_REQUIRED_NOT_LOAD, FUNC_DATA_REQUIRED_FILTEROUT, + FUNC_DATA_REQUIRED_ALL_FILTEROUT, } EFuncDataRequired; EFuncDataRequired fmFuncDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow); diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 6a1709c6f7..a8b2a74019 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -183,7 +183,7 @@ typedef struct STsdbReader STsdbReader; int32_t tsdbSetTableList(STsdbReader *pReader, const void *pTableList, int32_t num); int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableList, int32_t numOfTables, - SSDataBlock *pResBlock, STsdbReader **ppReader, const char *idstr, bool countOnly); + SSDataBlock *pResBlock, STsdbReader **ppReader, const char *idstr, bool countOnly, SHashObj** pIgnoreTables); void tsdbReaderSetId(STsdbReader* pReader, const char* idstr); void tsdbReaderClose(STsdbReader *pReader); diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 5bd41dd86f..2b6f175158 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -193,6 +193,7 @@ struct STsdbReader { SBlockInfoBuf blockInfoBuf; int32_t step; STsdbReader* innerReader[2]; + SHashObj** pIgnoreTables; }; static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter); @@ -2704,15 +2705,21 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) { int64_t st = taosGetTimestampUs(); int32_t step = asc ? 1 : -1; double el = 0; + SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter); + SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; STableBlockScanInfo* pBlockScanInfo = NULL; if (pBlockInfo != NULL) { + if (*pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) { + setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order); + return code; + } + pBlockScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBlockInfo->uid, pReader->idStr); if (pBlockScanInfo == NULL) { goto _end; } - SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter); TSDBKEY keyInBuf = getCurrentKeyInBuf(pBlockScanInfo, pReader); // it is a clean block, load it directly @@ -2731,9 +2738,12 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) { } } else { // file blocks not exist pBlockScanInfo = *pReader->status.pTableIter; + if (*pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockScanInfo->uid, sizeof(pBlockScanInfo->uid))) { + setBlockAllDumped(pDumpInfo, pBlock->maxKey.ts, pReader->order); + return code; + } } - SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SBlockData* pBlockData = &pReader->status.fileBlockData; while (1) { @@ -3009,6 +3019,14 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) { while (1) { // load the last data block of current table STableBlockScanInfo* pScanInfo = *(STableBlockScanInfo**)pStatus->pTableIter; + if (*pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pScanInfo->uid, sizeof(pScanInfo->uid))) { + bool hasNexTable = moveToNextTable(pUidList, pStatus); + if (!hasNexTable) { + return TSDB_CODE_SUCCESS; + } + + continue; + } bool hasVal = initLastBlockReader(pLastBlockReader, pScanInfo, pReader); if (!hasVal) { @@ -3060,20 +3078,24 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) { static int32_t doBuildDataBlock(STsdbReader* pReader) { int32_t code = TSDB_CODE_SUCCESS; - SDataBlk* pBlock = NULL; SReaderStatus* pStatus = &pReader->status; SDataBlockIter* pBlockIter = &pStatus->blockIter; STableBlockScanInfo* pScanInfo = NULL; SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter); SLastBlockReader* pLastBlockReader = pReader->status.fileIter.pLastBlockReader; + SDataBlk* pBlock = getCurrentBlock(pBlockIter); + + if (*pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pBlockInfo->uid, sizeof(pBlockInfo->uid))) { + setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlock->maxKey.ts, pReader->order); + return code; + } pScanInfo = getTableBlockScanInfo(pReader->status.pTableMap, pBlockInfo->uid, pReader->idStr); if (pScanInfo == NULL) { return terrno; } - pBlock = getCurrentBlock(pBlockIter); initLastBlockReader(pLastBlockReader, pScanInfo, pReader); TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader); @@ -3309,6 +3331,13 @@ static int32_t buildBlockFromBufferSequentially(STsdbReader* pReader) { // } STableBlockScanInfo** pBlockScanInfo = pStatus->pTableIter; + if (*pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &(*pBlockScanInfo)->uid, sizeof((*pBlockScanInfo)->uid))) { + bool hasNexTable = moveToNextTable(pUidList, pStatus); + if (!hasNexTable) { + return TSDB_CODE_SUCCESS; + } + } + initMemDataIterator(*pBlockScanInfo, pReader); int64_t endKey = (ASCENDING_TRAVERSE(pReader->order)) ? INT64_MAX : INT64_MIN; @@ -4257,7 +4286,7 @@ static void freeSchemaFunc(void* param) { // ====================================== EXPOSED APIs ====================================== int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableList, int32_t numOfTables, - SSDataBlock* pResBlock, STsdbReader** ppReader, const char* idstr, bool countOnly) { + SSDataBlock* pResBlock, STsdbReader** ppReader, const char* idstr, bool countOnly, SHashObj** pIgnoreTables) { STimeWindow window = pCond->twindows; int32_t capacity = pVnode->config.tsdbCfg.maxRows; @@ -4357,6 +4386,8 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL pReader->readMode = READ_MODE_COUNT_ONLY; } + pReader->pIgnoreTables = pIgnoreTables; + tsdbDebug("%p total numOfTable:%d in this query %s", pReader, numOfTables, pReader->idStr); return code; diff --git a/source/libs/executor/inc/executil.h b/source/libs/executor/inc/executil.h index 2e92f9e396..a4a55e91fb 100644 --- a/source/libs/executor/inc/executil.h +++ b/source/libs/executor/inc/executil.h @@ -89,7 +89,25 @@ typedef struct SColMatchInfo { } SColMatchInfo; typedef struct SExecTaskInfo SExecTaskInfo; -typedef struct STableListInfo STableListInfo; + + +typedef struct STableListIdInfo { + uint64_t suid; + uint64_t uid; + int32_t tableType; +} STableListIdInfo; + +// If the numOfOutputGroups is 1, the data blocks that belongs to different groups will be provided randomly +// The numOfOutputGroups is specified by physical plan. and will not be affect by numOfGroups +typedef struct STableListInfo { + bool oneTableForEachGroup; + int32_t numOfOuputGroups; // the data block will be generated one by one + int32_t* groupOffset; // keep the offset value for each group in the tableList + SArray* pTableList; + SHashObj* map; // speedup acquire the tableQueryInfo by table uid + STableListIdInfo idInfo; // this maybe the super table or ordinary table +} STableListInfo; + struct SqlFunctionCtx; int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle, diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 85424fd7de..f4e0108b92 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -327,6 +327,7 @@ typedef struct STableScanInfo { SScanInfo scanInfo; int32_t scanTimes; SSDataBlock* pResBlock; + SHashObj* pIgnoreTables; SSampleExecInfo sample; // sample execution info int32_t currentGroupId; int32_t currentTable; diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 3a8875e997..c265c0b003 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -27,23 +27,6 @@ #include "executorimpl.h" #include "tcompression.h" -typedef struct STableListIdInfo { - uint64_t suid; - uint64_t uid; - int32_t tableType; -} STableListIdInfo; - -// If the numOfOutputGroups is 1, the data blocks that belongs to different groups will be provided randomly -// The numOfOutputGroups is specified by physical plan. and will not be affect by numOfGroups -struct STableListInfo { - bool oneTableForEachGroup; - int32_t numOfOuputGroups; // the data block will be generated one by one - int32_t* groupOffset; // keep the offset value for each group in the tableList - SArray* pTableList; - SHashObj* map; // speedup acquire the tableQueryInfo by table uid - STableListIdInfo idInfo; // this maybe the super table or ordinary table -}; - typedef struct tagFilterAssist { SHashObj* colHash; int32_t index; diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index f979987ad0..c6f5d3912d 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -1170,7 +1170,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT if (pScanBaseInfo->dataReader == NULL) { int32_t code = tsdbReaderOpen(pScanBaseInfo->readHandle.vnode, &pScanBaseInfo->cond, &keyInfo, 1, - pScanInfo->pResBlock, &pScanBaseInfo->dataReader, id, false); + pScanInfo->pResBlock, &pScanBaseInfo->dataReader, id, false, NULL); if (code != TSDB_CODE_SUCCESS) { qError("prepare read tsdb snapshot failed, uid:%" PRId64 ", code:%s %s", pOffset->uid, tstrerror(code), id); terrno = code; @@ -1229,7 +1229,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT int32_t size = tableListGetSize(pTableListInfo); tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, NULL, &pInfo->dataReader, NULL, - false); + false, NULL); cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond); strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 2389c7252e..1c57db6d71 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -192,9 +192,24 @@ static SResultRow* getTableGroupOutputBuf(SOperatorInfo* pOperator, uint64_t gro return (SResultRow*)((char*)(*pPage) + p1->offset); } +static int32_t insertTableToScanIgnoreList(STableScanInfo* pTableScanInfo, uint64_t uid) { + if (NULL == pTableScanInfo->pIgnoreTables) { + int32_t tableNum = taosArrayGetSize(pTableScanInfo->base.pTableListInfo->pTableList); + pTableScanInfo->pIgnoreTables = taosHashInit(tableNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK); + if (NULL == pTableScanInfo->pIgnoreTables) { + return TSDB_CODE_OUT_OF_MEMORY; + } + } + + taosHashPut(pTableScanInfo->pIgnoreTables, &uid, sizeof(uid), &pTableScanInfo->scanTimes, sizeof(pTableScanInfo->scanTimes)); + + return TSDB_CODE_SUCCESS; +} + static int32_t doDynamicPruneDataBlock(SOperatorInfo* pOperator, SDataBlockInfo* pBlockInfo, uint32_t* status) { STableScanInfo* pTableScanInfo = pOperator->info; - + int32_t code = TSDB_CODE_SUCCESS; + if (pTableScanInfo->base.pdInfo.pExprSup == NULL) { return TSDB_CODE_SUCCESS; } @@ -226,9 +241,10 @@ static int32_t doDynamicPruneDataBlock(SOperatorInfo* pOperator, SDataBlockInfo* if (notLoadBlock) { *status = FUNC_DATA_REQUIRED_NOT_LOAD; + code = insertTableToScanIgnoreList(pTableScanInfo, pBlockInfo->id.uid); } - return TSDB_CODE_SUCCESS; + return code; } static bool doFilterByBlockSMA(SFilterInfo* pFilterInfo, SColumnDataAgg** pColsAgg, int32_t numOfCols, @@ -380,7 +396,13 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanBase* pTableSca GET_TASKID(pTaskInfo), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows); pCost->skipBlocks += 1; tsdbReleaseDataBlock(pTableScanInfo->dataReader); - *status = FUNC_DATA_REQUIRED_FILTEROUT; + + STableScanInfo* pTableScanInfo = pOperator->info; + if (taosHashGetSize(pTableScanInfo->pIgnoreTables) == taosArrayGetSize(pTableScanInfo->base.pTableListInfo->pTableList)) { + *status = FUNC_DATA_REQUIRED_ALL_FILTEROUT; + } else { + *status = FUNC_DATA_REQUIRED_FILTEROUT; + } return TSDB_CODE_SUCCESS; } @@ -695,6 +717,10 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) { T_LONG_JMP(pTaskInfo->env, code); } + if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) { + break; + } + // current block is filter out according to filter condition, continue load the next block if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) { continue; @@ -734,6 +760,7 @@ static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) { } pTableScanInfo->scanTimes += 1; + taosHashClear(pTableScanInfo->pIgnoreTables); if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) { setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); @@ -761,6 +788,7 @@ static SSDataBlock* doGroupedTableScan(SOperatorInfo* pOperator) { } pTableScanInfo->scanTimes += 1; + taosHashClear(pTableScanInfo->pIgnoreTables); if (pTableScanInfo->scanTimes < total) { setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); @@ -825,7 +853,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { ASSERT(pInfo->base.dataReader == NULL); int32_t code = tsdbReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock, - (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), pInfo->countOnly); + (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), pInfo->countOnly, &pInfo->pIgnoreTables); if (code != TSDB_CODE_SUCCESS) { T_LONG_JMP(pTaskInfo->env, code); } @@ -894,6 +922,7 @@ static void destroyTableScanBase(STableScanBase* pBase) { static void destroyTableScanOperatorInfo(void* param) { STableScanInfo* pTableScanInfo = (STableScanInfo*)param; blockDataDestroy(pTableScanInfo->pResBlock); + taosHashCleanup(pTableScanInfo->pIgnoreTables); destroyTableScanBase(&pTableScanInfo->base); taosMemoryFreeClear(param); } @@ -1059,7 +1088,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU SSDataBlock* pBlock = pTableScanInfo->pResBlock; STsdbReader* pReader = NULL; int32_t code = tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock, - (STsdbReader**)&pReader, GET_TASKID(pTaskInfo), false); + (STsdbReader**)&pReader, GET_TASKID(pTaskInfo), false, NULL); if (code != TSDB_CODE_SUCCESS) { terrno = code; T_LONG_JMP(pTaskInfo->env, code); @@ -2662,7 +2691,7 @@ static SSDataBlock* getTableDataBlockImpl(void* param) { SReadHandle* pHandle = &pInfo->base.readHandle; if (NULL == source->dataReader || !source->multiReader) { - code = tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &source->dataReader, GET_TASKID(pTaskInfo), false); + code = tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &source->dataReader, GET_TASKID(pTaskInfo), false, NULL); if (code != 0) { T_LONG_JMP(pTaskInfo->env, code); } @@ -2710,6 +2739,10 @@ static SSDataBlock* getTableDataBlockImpl(void* param) { T_LONG_JMP(pTaskInfo->env, code); } + if (status == FUNC_DATA_REQUIRED_ALL_FILTEROUT) { + break; + } + // current block is filter out according to filter condition, continue load the next block if (status == FUNC_DATA_REQUIRED_FILTEROUT || pBlock->info.rows == 0) { continue; diff --git a/source/libs/executor/src/sysscanoperator.c b/source/libs/executor/src/sysscanoperator.c index 1abe678ac6..7dbf182d9a 100644 --- a/source/libs/executor/src/sysscanoperator.c +++ b/source/libs/executor/src/sysscanoperator.c @@ -2269,7 +2269,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi size_t num = tableListGetSize(pTableListInfo); void* pList = tableListGetInfo(pTableListInfo, 0); - code = tsdbReaderOpen(readHandle->vnode, &cond, pList, num, pInfo->pResBlock, &pInfo->pHandle, pTaskInfo->id.str, false); + code = tsdbReaderOpen(readHandle->vnode, &cond, pList, num, pInfo->pResBlock, &pInfo->pHandle, pTaskInfo->id.str, false, NULL); cleanupQueryTableDataCond(&cond); if (code != 0) { goto _error; -- GitLab