From ad3075db1eceb730f7d549c1fca3755f4046a0e5 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 3 May 2022 23:23:49 +0800 Subject: [PATCH 1/3] refactor: do some internal refactor. --- source/libs/executor/inc/executorimpl.h | 9 +- source/libs/executor/src/executorimpl.c | 21 +- source/libs/executor/src/groupoperator.c | 4 +- source/libs/executor/src/scanoperator.c | 257 ++++++++---------- source/libs/executor/src/timewindowoperator.c | 10 +- 5 files changed, 136 insertions(+), 165 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 6305022b27..2795109603 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -268,7 +268,7 @@ typedef struct SOperatorFpSet { typedef struct SOperatorInfo { uint8_t operatorType; - bool blockingOptr; // block operator or not + bool blocking; // block operator or not uint8_t status; // denote if current operator is completed int32_t numOfOutput; // number of columns of the current operator results char* name; // name, used to show the query execution plan @@ -341,9 +341,9 @@ typedef struct STableScanInfo { int64_t elapsedTime; int32_t prevGroupId; // previous table group id SScanInfo scanInfo; - int32_t current; - SNode* pFilterNode; // filter operator info - SqlFunctionCtx* pCtx; // next operator query context + int32_t scanTimes; + SNode* pFilterNode; // filter info, which is push down by optimizer + SqlFunctionCtx* pCtx; // which belongs to the direct upstream operator operator query context SResultRowInfo* pResultRowInfo; int32_t* rowCellInfoOffset; SExprInfo* pExpr; @@ -397,7 +397,6 @@ typedef struct SSysTableScanInfo { SArray* scanCols; // SArray scan column id list SName name; SSDataBlock* pRes; - int32_t capacity; int64_t numOfBlocks; // extract basic running information. SLoadRemoteDataInfo loadInfo; } SSysTableScanInfo; diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 98804d4f1a..a6cdc6ab1a 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -132,6 +132,7 @@ void doSetOperatorCompleted(SOperatorInfo* pOperator) { int32_t operatorDummyOpenFn(SOperatorInfo* pOperator) { OPTR_SET_OPENED(pOperator); + pOperator->cost.openCost = 0; return TSDB_CODE_SUCCESS; } @@ -3282,7 +3283,7 @@ SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock pOperator->name = "ExchangeOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE; - pOperator->blockingOptr = false; + pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; pOperator->numOfOutput = pBlock->info.numOfCols; @@ -3673,7 +3674,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t pOperator->name = "SortedMerge"; // pOperator->operatorType = OP_SortedMerge; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; pOperator->numOfOutput = num; @@ -3756,7 +3757,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pR pOperator->name = "SortOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; @@ -4419,7 +4420,7 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->name = "TableAggregate"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_AGG; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; pOperator->pExpr = pExprInfo; @@ -4532,7 +4533,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* p pOperator->name = "ProjectOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PROJECT; - pOperator->blockingOptr = false; + pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->info = pInfo; pOperator->pExpr = pExprInfo; @@ -4615,7 +4616,7 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExp } pOperator->name = "FillOperator"; - pOperator->blockingOptr = false; + pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL; pOperator->pExpr = pExpr; @@ -4861,7 +4862,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo SSDataBlock* pResBlock = createResDataBlock(pScanPhyNode->node.pOutputDataBlockDesc); SQueryTableDataCond cond = {0}; - int32_t code = initQueryTableDataCond(&cond, pTableScanNode); + + int32_t code = initQueryTableDataCond(&cond, pTableScanNode); if (code != TSDB_CODE_SUCCESS) { return NULL; } @@ -4877,8 +4879,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) { SScanPhysiNode* pScanPhyNode = (SScanPhysiNode*)pPhyNode; // simple child table. - int32_t code = doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo, - queryId, taskId); + int32_t code = doCreateTableGroup(pHandle->meta, pScanPhyNode->tableType, pScanPhyNode->uid, pTableGroupInfo, queryId, taskId); SArray* tableIdList = extractTableIdList(pTableGroupInfo); SSDataBlock* pResBlock = createResDataBlock(pScanPhyNode->node.pOutputDataBlockDesc); @@ -5628,7 +5629,7 @@ SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOf pOperator->name = "JoinOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_JOIN; - pOperator->blockingOptr = false; + pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExprInfo; pOperator->numOfOutput = numOfCols; diff --git a/source/libs/executor/src/groupoperator.c b/source/libs/executor/src/groupoperator.c index acef8e4bc4..f5dd67217e 100644 --- a/source/libs/executor/src/groupoperator.c +++ b/source/libs/executor/src/groupoperator.c @@ -353,7 +353,7 @@ SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx initResultRowInfo(&pInfo->binfo.resultRowInfo, 8); pOperator->name = "GroupbyAggOperator"; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; // pOperator->operatorType = OP_Groupby; pOperator->pExpr = pExprInfo; @@ -612,7 +612,7 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SExprInfo* } pOperator->name = "PartitionOperator"; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_PARTITION; pInfo->binfo.pRes = pResultBlock; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index a66d2a263c..30865b42f5 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -13,7 +13,7 @@ * along with this program. If not, see . */ -#include "common/ttime.h" +#include "ttime.h" #include "filter.h" #include "function.h" #include "functionMgt.h" @@ -36,10 +36,10 @@ #define SET_REVERSE_SCAN_FLAG(_info) ((_info)->scanFlag = REVERSE_SCAN) #define SWITCH_ORDER(n) (((n) = ((n) == TSDB_ORDER_ASC) ? TSDB_ORDER_DESC : TSDB_ORDER_ASC)) -int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo); +static int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity); +static int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size, const char* dbName); -int32_t buildDbTableInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size, const char* dbName); -void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) { +static void switchCtxOrder(SqlFunctionCtx* pCtx, int32_t numOfOutput) { for (int32_t i = 0; i < numOfOutput; ++i) { SWITCH_ORDER(pCtx[i].order); } @@ -158,7 +158,7 @@ static bool overlapWithTimeWindow(SInterval* pInterval, SDataBlockInfo* pBlockIn return false; } -int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) { +static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; STableScanInfo* pInfo = pOperator->info; @@ -296,15 +296,16 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { return NULL; } - while (pTableScanInfo->current < pTableScanInfo->scanInfo.numOfAsc) { + // do the ascending order traverse in the first place. + while (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) { SSDataBlock* p = doTableScanImpl(pOperator); if (p != NULL) { return p; } - pTableScanInfo->current += 1; + pTableScanInfo->scanTimes += 1; - if (pTableScanInfo->current < pTableScanInfo->scanInfo.numOfAsc) { + if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) { setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); pTableScanInfo->scanFlag = REPEAT_SCAN; @@ -319,7 +320,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { } int32_t total = pTableScanInfo->scanInfo.numOfAsc + pTableScanInfo->scanInfo.numOfDesc; - if (pTableScanInfo->current < total) { + if (pTableScanInfo->scanTimes < total) { if (pTableScanInfo->cond.order == TSDB_ORDER_ASC) { prepareForDescendingScan(pTableScanInfo, pTableScanInfo->pCtx, pTableScanInfo->numOfOutput); tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond); @@ -329,21 +330,20 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { qDebug("%s start to descending order scan data blocks due to query func required, qrange:%" PRId64 "-%" PRId64, GET_TASKID(pTaskInfo), pWin->skey, pWin->ekey); - while (pTableScanInfo->current < total) { + while (pTableScanInfo->scanTimes < total) { SSDataBlock* p = doTableScanImpl(pOperator); if (p != NULL) { return p; } - pTableScanInfo->current += 1; + pTableScanInfo->scanTimes += 1; - if (pTableScanInfo->current < pTableScanInfo->scanInfo.numOfAsc) { + if (pTableScanInfo->scanTimes < pTableScanInfo->scanInfo.numOfAsc) { setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED); pTableScanInfo->scanFlag = REPEAT_SCAN; qDebug("%s start to repeat descending order scan data blocks due to query func required, qrange:%" PRId64 - "-%" PRId64, - GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey); + "-%" PRId64, GET_TASKID(pTaskInfo), pTaskInfo->window.skey, pTaskInfo->window.ekey); // do prepare for the next round table scan operation tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond); @@ -372,48 +372,50 @@ SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, SQueryTableDataCon pInfo->cond = *pCond; pInfo->scanInfo = (SScanInfo){.numOfAsc = scanInfo[0], .numOfDesc = scanInfo[1]}; - pInfo->interval = *pInterval; - pInfo->sampleRatio = sampleRatio; + pInfo->interval = *pInterval; + pInfo->sampleRatio = sampleRatio; pInfo->dataBlockLoadFlag = dataLoadFlag; - pInfo->pResBlock = pResBlock; - pInfo->pFilterNode = pCondition; - pInfo->dataReader = pDataReader; - pInfo->current = 0; - pInfo->scanFlag = MAIN_SCAN; - pInfo->pColMatchInfo = pColMatchInfo; + pInfo->pResBlock = pResBlock; + pInfo->pFilterNode = pCondition; + pInfo->dataReader = pDataReader; + pInfo->scanFlag = MAIN_SCAN; + pInfo->pColMatchInfo = pColMatchInfo; - pOperator->name = "TableScanOperator"; - pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN; - pOperator->blockingOptr = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->numOfOutput = numOfOutput; - pOperator->fpSet.getNextFn = doTableScan; - pOperator->pTaskInfo = pTaskInfo; + pOperator->name = "TableScanOperator"; // for dubug purpose + pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->numOfOutput = numOfOutput; + pOperator->pTaskInfo = pTaskInfo; + + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScan, NULL, NULL, NULL, NULL, NULL, NULL); static int32_t cost = 0; - pOperator->cost.openCost = ++cost; + + // for non-blocking operator, the open cost is always 0 + pOperator->cost.openCost = 0; pOperator->cost.totalCost = ++cost; pOperator->resultInfo.totalRows = ++cost; return pOperator; } -SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle) { +SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* pTaskInfo) { STableScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableScanInfo)); - - pInfo->dataReader = pTsdbReadHandle; - pInfo->current = 0; - pInfo->prevGroupId = -1; - SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); - pOperator->name = "TableSeqScanOperator"; - pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN; - pOperator->blockingOptr = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->fpSet.getNextFn = doTableScanImpl; + pInfo->dataReader = pReadHandle; + pInfo->prevGroupId = -1; + + pOperator->name = "TableSeqScanOperator"; + pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->pTaskInfo = pTaskInfo; + + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableScanImpl, NULL, NULL, NULL, NULL, NULL, NULL); return pOperator; } @@ -482,7 +484,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(void* dataReader, SExecTaskInfo* pOperator->name = "DataBlockInfoScanOperator"; // pOperator->operatorType = OP_TableBlockInfoScan; - pOperator->blockingOptr = false; + pOperator->blocking = false; pOperator->status = OP_NOT_OPENED; pOperator->fpSet._openFn = operatorDummyOpenFn; pOperator->fpSet.getNextFn = doBlockInfoScan; @@ -636,20 +638,22 @@ SOperatorInfo* createStreamScanOperatorInfo(void* streamReadHandle, SSDataBlock* } pInfo->readerHandle = streamReadHandle; - pInfo->pRes = pResBlock; - pInfo->pCondition = pCondition; + pInfo->pRes = pResBlock; + pInfo->pCondition = pCondition; - pOperator->name = "StreamBlockScanOperator"; - pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; - pOperator->blockingOptr = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->numOfOutput = pResBlock->info.numOfCols; - pOperator->fpSet._openFn = operatorDummyOpenFn; + pOperator->name = "StreamBlockScanOperator"; + pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->numOfOutput = pResBlock->info.numOfCols; + pOperator->fpSet._openFn = operatorDummyOpenFn; pOperator->fpSet.getNextFn = doStreamBlockScan; - pOperator->fpSet.closeFn = operatorDummyCloseFn; + pOperator->fpSet.closeFn = operatorDummyCloseFn; pOperator->pTaskInfo = pTaskInfo; + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamBlockScan, NULL, NULL, operatorDummyCloseFn, NULL, NULL, NULL); + return pOperator; } @@ -671,14 +675,12 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) { switch (nType) { case QUERY_NODE_OPERATOR: { SOperatorNode* node = (SOperatorNode*)pNode; - if (OP_TYPE_EQUAL == node->opType) { *(int32_t*)pContext = 1; return DEAL_RES_CONTINUE; } *(int32_t*)pContext = 0; - return DEAL_RES_IGNORE_CHILD; } case QUERY_NODE_COLUMN: { @@ -709,19 +711,17 @@ EDealRes getDBNameFromConditionWalker(SNode* pNode, void* pContext) { default: break; } - return DEAL_RES_CONTINUE; } -void getDBNameFromCondition(SNode* pCondition, char* dbName) { +static void getDBNameFromCondition(SNode* pCondition, const char* dbName) { if (NULL == pCondition) { return; } - - nodesWalkExpr(pCondition, getDBNameFromConditionWalker, dbName); + nodesWalkExpr(pCondition, getDBNameFromConditionWalker, (char*) dbName); } -static int32_t loadSysTableContentCb(void* param, const SDataBuf* pMsg, int32_t code) { +static int32_t loadSysTableCallback(void* param, const SDataBuf* pMsg, int32_t code) { SOperatorInfo* operator=(SOperatorInfo*) param; SSysTableScanInfo* pScanResInfo = (SSysTableScanInfo*)operator->info; if (TSDB_CODE_SUCCESS == code) { @@ -746,13 +746,14 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) { } SFilterInfo* filter = NULL; - int32_t code = filterInitFromNode(pInfo->pCondition, &filter, 0); + + int32_t code = filterInitFromNode(pInfo->pCondition, &filter, 0); SFilterColumnParam param1 = {.numOfCols = pInfo->pRes->info.numOfCols, .pDataBlock = pInfo->pRes->pDataBlock}; code = filterSetDataFromSlotId(filter, ¶m1); int8_t* rowRes = NULL; - bool keep = filterExecute(filter, pInfo->pRes, &rowRes, NULL, param1.numOfCols); + bool keep = filterExecute(filter, pInfo->pRes, &rowRes, NULL, param1.numOfCols); filterFreeInfo(filter); SSDataBlock* px = createOneDataBlock(pInfo->pRes, false); @@ -795,61 +796,31 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) { static SSDataBlock* buildSysTableMetaBlock() { SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock)); - pBlock->info.numOfCols = 10; - pBlock->info.hasVarCol = true; + + size_t size = 0; + const SSysTableMeta *pMeta = NULL; + getInfosDbMeta(&pMeta, &size); + + int32_t index = 0; + for(int32_t i = 0; i < size; ++i) { + if(strcmp(pMeta[i].name, TSDB_INS_TABLE_USER_TABLES) == 0) { + index = i; + break; + } + } pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData)); - SColumnInfoData colInfoData = {0}; - colInfoData.info.colId = 1; - colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR; - colInfoData.info.bytes = (TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE; - taosArrayPush(pBlock->pDataBlock, &colInfoData); + for(int32_t i = 0; i < pMeta[index].colNum; ++i) { + SColumnInfoData colInfoData = {0}; + colInfoData.info.colId = i + 1; + colInfoData.info.type = pMeta[index].schema[i].type; + colInfoData.info.bytes = pMeta[index].schema[i].bytes; + taosArrayPush(pBlock->pDataBlock, &colInfoData); + } - colInfoData.info.colId = 2; - colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR; - colInfoData.info.bytes = (TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 3; - colInfoData.info.type = TSDB_DATA_TYPE_TIMESTAMP; - colInfoData.info.bytes = 8; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 4; - colInfoData.info.type = TSDB_DATA_TYPE_INT; - colInfoData.info.bytes = 4; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 5; - colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR; - colInfoData.info.bytes = (TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 6; - colInfoData.info.type = TSDB_DATA_TYPE_BIGINT; - colInfoData.info.bytes = 8; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 7; - colInfoData.info.type = TSDB_DATA_TYPE_INT; - colInfoData.info.bytes = 4; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 8; - colInfoData.info.type = TSDB_DATA_TYPE_INT; - colInfoData.info.bytes = 4; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 9; - colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR; - colInfoData.info.bytes = 512 + VARSTR_HEADER_SIZE; - taosArrayPush(pBlock->pDataBlock, &colInfoData); - - colInfoData.info.colId = 10; - colInfoData.info.type = TSDB_DATA_TYPE_VARCHAR; - colInfoData.info.bytes = 20 + VARSTR_HEADER_SIZE; - taosArrayPush(pBlock->pDataBlock, &colInfoData); + pBlock->info.numOfCols = pMeta[index].colNum; + pBlock->info.hasVarCol = true; return pBlock; } @@ -868,7 +839,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { return NULL; } - buildSysDbTableInfo(pInfo); + buildSysDbTableInfo(pInfo, pOperator->resultInfo.capacity); doFilterResult(pInfo); pInfo->loadInfo.totalRows += pInfo->pRes->info.rows; @@ -896,7 +867,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { varDataSetLen(dbname, strlen(varDataVal(dbname))); SSDataBlock* p = buildSysTableMetaBlock(); - blockDataEnsureCapacity(p, pInfo->capacity); + blockDataEnsureCapacity(p, pOperator->resultInfo.capacity); char n[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; while (metaTbCursorNext(pInfo->pCur) == 0) { @@ -977,7 +948,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { pColInfoData = taosArrayGet(p->pDataBlock, 9); colDataAppend(pColInfoData, numOfRows, str, false); - if (++numOfRows >= pInfo->capacity) { + if (++numOfRows >= pOperator->resultInfo.capacity) { break; } } @@ -1022,7 +993,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { pMsgSendInfo->msgInfo.pData = buf1; pMsgSendInfo->msgInfo.len = contLen; pMsgSendInfo->msgType = TDMT_MND_SYSTABLE_RETRIEVE; - pMsgSendInfo->fp = loadSysTableContentCb; + pMsgSendInfo->fp = loadSysTableCallback; int64_t transporterId = 0; int32_t code = asyncSendMsgToServer(pInfo->pTransporter, &pInfo->epSet, &transporterId, pMsgSendInfo); @@ -1060,9 +1031,9 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { } } -int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo) { +int32_t buildSysDbTableInfo(const SSysTableScanInfo* pInfo, int32_t capacity) { SSDataBlock* p = buildSysTableMetaBlock(); - blockDataEnsureCapacity(p, pInfo->capacity); + blockDataEnsureCapacity(p, capacity); size_t size = 0; const SSysTableMeta* pSysDbTableMeta = NULL; @@ -1133,18 +1104,19 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe return NULL; } - pInfo->accountId = accountId; + pInfo->accountId = accountId; pInfo->showRewrite = showRewrite; - pInfo->pRes = pResBlock; - pInfo->capacity = 4096; - pInfo->pCondition = pCondition; - pInfo->scanCols = colList; + pInfo->pRes = pResBlock; + pInfo->pCondition = pCondition; + pInfo->scanCols = colList; + + initResultSizeInfo(pOperator, 4096); tNameAssign(&pInfo->name, pName); const char* name = tNameGetTableName(&pInfo->name); if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) { pInfo->readHandle = *(SReadHandle*) readHandle; - blockDataEnsureCapacity(pInfo->pRes, pInfo->capacity); + blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); } else { tsem_init(&pInfo->ready, 0, 0); pInfo->epSet = epset; @@ -1173,12 +1145,12 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe #endif } - pOperator->name = "SysTableScanOperator"; + pOperator->name = "SysTableScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN; - pOperator->blockingOptr = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; - pOperator->numOfOutput = pResBlock->info.numOfCols; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->numOfOutput = pResBlock->info.numOfCols; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSysTableScan, NULL, NULL, destroySysScanOperator, NULL, NULL, NULL); pOperator->pTaskInfo = pTaskInfo; @@ -1188,7 +1160,6 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSDataBlock* pRe static SSDataBlock* doTagScan(SOperatorInfo* pOperator) { #if 0 - SOperatorInfo* pOperator = (SOperatorInfo*) param; if (pOperator->status == OP_EXEC_DONE) { return NULL; } @@ -1322,29 +1293,29 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) { pInfo->pRes = blockDataDestroy(pInfo->pRes); } -SOperatorInfo* createTagScanOperatorInfo(void* pReaderHandle, SExprInfo* pExpr, int32_t numOfOutput, - SExecTaskInfo* pTaskInfo) { +SOperatorInfo* createTagScanOperatorInfo(void* readHandle, SExprInfo* pExpr, int32_t numOfOutput, SExecTaskInfo* pTaskInfo) { STagScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); if (pInfo == NULL || pOperator == NULL) { goto _error; } - pInfo->pReader = pReaderHandle; - pInfo->curPos = 0; - pOperator->name = "TagScanOperator"; + pInfo->pReader = readHandle; + pInfo->curPos = 0; + pOperator->name = "TagScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN; - pOperator->blockingOptr = false; - pOperator->status = OP_NOT_OPENED; - pOperator->info = pInfo; + pOperator->blocking = false; + pOperator->status = OP_NOT_OPENED; + pOperator->info = pInfo; + pOperator->pExpr = pExpr; + pOperator->numOfOutput = numOfOutput; + pOperator->pTaskInfo = pTaskInfo; pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTagScan, NULL, NULL, destroyTagScanOperatorInfo, NULL, NULL, NULL); - pOperator->pExpr = pExpr; - pOperator->numOfOutput = numOfOutput; - pOperator->pTaskInfo = pTaskInfo; return pOperator; + _error: taosMemoryFree(pInfo); taosMemoryFree(pOperator); diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 33112b5982..0384f52dda 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1078,7 +1078,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->name = "TimeIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERVAL; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExprInfo; pOperator->pTaskInfo = pTaskInfo; @@ -1137,7 +1137,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr pOperator->name = "StreamTimeIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERVAL; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExprInfo; pOperator->pTaskInfo = pTaskInfo; @@ -1343,7 +1343,7 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->name = "TimeSliceOperator"; // pOperator->operatorType = OP_AllTimeWindow; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExprInfo; pOperator->numOfOutput = numOfCols; @@ -1385,7 +1385,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf pInfo->tsSlotId = tsSlotId; pOperator->name = "StateWindowOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExpr; pOperator->numOfOutput = numOfCols; @@ -1437,7 +1437,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo pInfo->reptScan = false; pOperator->name = "SessionWindowAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW; - pOperator->blockingOptr = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExprInfo; pOperator->numOfOutput = numOfCols; From 05bd6a52c2705c5078c3575a4dd40253353d1bc5 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Tue, 3 May 2022 23:52:17 +0800 Subject: [PATCH 2/3] refactor: do some internal refactor. --- source/libs/executor/inc/executorimpl.h | 28 ++++++++++++++----------- source/libs/executor/src/executorimpl.c | 13 +++++++----- source/libs/executor/src/scanoperator.c | 11 +++++----- 3 files changed, 29 insertions(+), 23 deletions(-) diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 2795109603..0404b4d447 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -119,6 +119,17 @@ typedef struct SLimit { int64_t offset; } SLimit; +typedef struct SFileBlockLoadRecorder { + uint64_t totalRows; + uint64_t totalCheckedRows; + uint32_t totalBlocks; + uint32_t loadBlocks; + uint32_t loadBlockStatis; + uint32_t skipBlocks; + uint32_t filterOutBlocks; + uint64_t elapsedTime; +} SFileBlockLoadRecorder; + typedef struct STaskCostInfo { int64_t created; int64_t start; @@ -132,14 +143,10 @@ typedef struct STaskCostInfo { uint64_t loadDataInCacheSize; uint64_t loadDataTime; - uint64_t totalRows; - uint64_t totalCheckedRows; - uint32_t totalBlocks; - uint32_t loadBlocks; - uint32_t loadBlockStatis; - uint32_t skipBlocks; - uint32_t filterOutBlocks; + + SFileBlockLoadRecorder* pRecoder; uint64_t elapsedTime; + uint64_t firstStageMergeTime; uint64_t winInfoSize; uint64_t tableInfoSize; @@ -333,13 +340,10 @@ typedef struct SScanInfo { typedef struct STableScanInfo { void* dataReader; - - int32_t numOfBlocks; // extract basic running information. - int32_t numOfSkipped; - int32_t numOfBlockStatis; + SFileBlockLoadRecorder readRecorder; int64_t numOfRows; int64_t elapsedTime; - int32_t prevGroupId; // previous table group id +// int32_t prevGroupId; // previous table group id SScanInfo scanInfo; int32_t scanTimes; SNode* pFilterNode; // filter info, which is push down by optimizer diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index a6cdc6ab1a..a79abc9949 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -1593,8 +1593,8 @@ int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableSc STaskCostInfo* pCost = &pTaskInfo->cost; - pCost->totalBlocks += 1; - pCost->totalRows += pBlock->info.rows; +// pCost->totalBlocks += 1; +// pCost->totalRows += pBlock->info.rows; #if 0 // Calculate all time windows that are overlapping or contain current data block. // If current data block is contained by all possible time window, do not load current data block. @@ -2415,8 +2415,8 @@ void queryCostStatis(SExecTaskInfo* pTaskInfo) { qDebug("%s :cost summary: elapsed time:%" PRId64 " us, first merge:%" PRId64 " us, total blocks:%d, " "load block statis:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64, - GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pSummary->totalBlocks, - pSummary->loadBlockStatis, pSummary->loadBlocks, pSummary->totalRows, pSummary->totalCheckedRows); + GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pSummary->pRecoder->totalBlocks, + pSummary->pRecoder->loadBlockStatis, pSummary->pRecoder->loadBlocks, pSummary->pRecoder->totalRows, pSummary->pRecoder->totalCheckedRows); // // qDebug("QInfo:0x%"PRIx64" :cost summary: winResPool size:%.2f Kb, numOfWin:%"PRId64", tableInfoSize:%.2f Kb, // hashTable:%.2f Kb", pQInfo->qId, pSummary->winInfoSize/1024.0, @@ -4869,9 +4869,12 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo } SInterval interval = extractIntervalInfo(pTableScanNode); - return createTableScanOperatorInfo(pDataReader, &cond, numOfCols, pTableScanNode->dataRequired, + SOperatorInfo* pOperator = createTableScanOperatorInfo(pDataReader, &cond, numOfCols, pTableScanNode->dataRequired, pTableScanNode->scanSeq, pColList, pResBlock, pScanPhyNode->node.pConditions, &interval, pTableScanNode->ratio, pTaskInfo); + STableScanInfo* pScanInfo = pOperator->info; + pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder; + return pOperator; } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) { SExchangePhysiNode* pExchange = (SExchangePhysiNode*)pPhyNode; SSDataBlock* pResBlock = createResDataBlock(pExchange->node.pOutputDataBlockDesc); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 30865b42f5..53b436c70d 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -162,7 +162,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; STableScanInfo* pInfo = pOperator->info; - STaskCostInfo* pCost = &pTaskInfo->cost; + SFileBlockLoadRecorder* pCost = &pTableScanInfo->readRecorder; pCost->totalBlocks += 1; pCost->totalRows += pBlock->info.rows; @@ -188,11 +188,11 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca } else if (*status == FUNC_DATA_REQUIRED_STATIS_LOAD) { pCost->loadBlockStatis += 1; - bool allHave = true; + bool allColumnsHaveAgg = true; SColumnDataAgg** pColAgg = NULL; - tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->dataReader, &pColAgg, &allHave); + tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->dataReader, &pColAgg, &allColumnsHaveAgg); - if (allHave == true) { + if (allColumnsHaveAgg == true) { int32_t numOfCols = pBlock->info.numOfCols; // todo create this buffer during creating operator @@ -266,7 +266,6 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator) { longjmp(pOperator->pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED); } - pTableScanInfo->numOfBlocks += 1; tsdbRetrieveDataBlockInfo(pTableScanInfo->dataReader, &pBlock->info); uint32_t status = 0; @@ -406,7 +405,7 @@ SOperatorInfo* createTableSeqScanOperatorInfo(void* pReadHandle, SExecTaskInfo* SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); pInfo->dataReader = pReadHandle; - pInfo->prevGroupId = -1; +// pInfo->prevGroupId = -1; pOperator->name = "TableSeqScanOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN; From db00d25b9b7f5d849122e56a62056b9f9fb36d90 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Wed, 4 May 2022 14:34:38 +0800 Subject: [PATCH 3/3] fix(query): add one more check for null pointer. --- source/libs/executor/src/executorimpl.c | 13 +++++++------ source/libs/executor/src/scanoperator.c | 1 + source/libs/executor/src/timewindowoperator.c | 2 +- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index a79abc9949..b5d972321b 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -2412,12 +2412,13 @@ void queryCostStatis(SExecTaskInfo* pTaskInfo) { // // calculateOperatorProfResults(pQInfo); - qDebug("%s :cost summary: elapsed time:%" PRId64 " us, first merge:%" PRId64 - " us, total blocks:%d, " - "load block statis:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64, - GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pSummary->pRecoder->totalBlocks, - pSummary->pRecoder->loadBlockStatis, pSummary->pRecoder->loadBlocks, pSummary->pRecoder->totalRows, pSummary->pRecoder->totalCheckedRows); - // + SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder; + if (pSummary->pRecoder != NULL) { + qDebug("%s :cost summary: elapsed time:%" PRId64 " us, first merge:%" PRId64 " us, total blocks:%d, " + "load block statis:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64, + GET_TASKID(pTaskInfo), pSummary->elapsedTime, pSummary->firstStageMergeTime, pRecorder->totalBlocks, + pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows, pRecorder->totalCheckedRows); + } // qDebug("QInfo:0x%"PRIx64" :cost summary: winResPool size:%.2f Kb, numOfWin:%"PRId64", tableInfoSize:%.2f Kb, // hashTable:%.2f Kb", pQInfo->qId, pSummary->winInfoSize/1024.0, // pSummary->numOfTimeWindows, pSummary->tableInfoSize/1024.0, pSummary->hashSize/1024.0); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 53b436c70d..7e721455c3 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -236,6 +236,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca } relocateColumnData(pBlock, pTableScanInfo->pColMatchInfo, pCols); + // todo record the filter time cost doFilter(pTableScanInfo->pFilterNode, pBlock); if (pBlock->info.rows == 0) { pCost->filterOutBlocks += 1; diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 0384f52dda..580614dc02 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1078,7 +1078,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pOperator->name = "TimeIntervalAggOperator"; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERVAL; - pOperator->blocking = true; + pOperator->blocking = true; pOperator->status = OP_NOT_OPENED; pOperator->pExpr = pExprInfo; pOperator->pTaskInfo = pTaskInfo;