From 2ead89e3d6530857f41baba4d9171aa9debc0e46 Mon Sep 17 00:00:00 2001 From: shenglian zhou Date: Fri, 18 Nov 2022 11:53:33 +0800 Subject: [PATCH] feat: table merge scan save work --- include/libs/function/function.h | 1 + source/dnode/vnode/inc/vnode.h | 6 ++ source/dnode/vnode/src/inc/vnodeInt.h | 6 -- source/libs/executor/inc/executorimpl.h | 11 ++++ source/libs/executor/src/executil.c | 1 + source/libs/executor/src/executorimpl.c | 5 ++ source/libs/executor/src/scanoperator.c | 80 +++++++++++++++++++++++++ 7 files changed, 104 insertions(+), 6 deletions(-) diff --git a/include/libs/function/function.h b/include/libs/function/function.h index 6f2a675466..d6e584cc10 100644 --- a/include/libs/function/function.h +++ b/include/libs/function/function.h @@ -163,6 +163,7 @@ typedef struct tExprNode { int32_t functionId; int32_t num; struct SFunctionNode *pFunctNode; + int32_t functionType; } _function; struct { diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 370103c222..fcbe5e476b 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -109,6 +109,12 @@ int metaGetTableUidByName(void *meta, char *tbName, int64_t *uid); int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType); bool metaIsTableExist(SMeta *pMeta, tb_uid_t uid); +typedef struct { + int64_t uid; + int64_t ctbNum; +} SMetaStbStats; +int32_t metaGetStbStats(SMeta* pMeta, int64_t uid, SMetaStbStats* pInfo); + typedef struct SMetaFltParam { tb_uid_t suid; int16_t cid; diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index ac9fabf052..1ca80c7570 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -144,12 +144,6 @@ typedef struct SMetaInfo { } SMetaInfo; int32_t metaGetInfo(SMeta* pMeta, int64_t uid, SMetaInfo* pInfo); -typedef struct { - int64_t uid; - int64_t ctbNum; -} SMetaStbStats; -int32_t metaGetStbStats(SMeta* pMeta, int64_t uid, SMetaStbStats* pInfo); - // tsdb int tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeepCfg* pKeepCfg, int8_t rollback); int tsdbClose(STsdb** pTsdb); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index b0da277cfb..6aa2a1e5d5 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -549,6 +549,17 @@ typedef struct SSysTableScanInfo { SLoadRemoteDataInfo loadInfo; } SSysTableScanInfo; +typedef struct STableCountScanInfo { + SReadHandle readHandle; + SSDataBlock* pRes; + SExprSupp pseudoSup; + SNode* pCondition; + SName name; + uint64_t suid; + uint64_t uid; + int8_t tableType; +} STableCountScanInfo; + typedef struct SBlockDistInfo { SSDataBlock* pResBlock; STsdbReader* pHandle; diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index d1046ff02c..d9b7682e45 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -1283,6 +1283,7 @@ void createExprFromOneNode(SExprInfo* pExp, SNode* pNode, int16_t slotId) { pExprNode->_function.functionId = pFuncNode->funcId; pExprNode->_function.pFunctNode = pFuncNode; + pExprNode->_function.functionType = pFuncNode->funcType; tstrncpy(pExprNode->_function.functionName, pFuncNode->functionName, tListLen(pExprNode->_function.functionName)); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 39e876800f..b1fe016d13 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -2612,6 +2612,8 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPT SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode); +SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* handle, STableCountScanPhysiNode* pNode, + SExecTaskInfo* pTaskInfo); int32_t extractTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, SExecTaskInfo* pTaskInfo) { SMetaReader mr = {0}; metaReaderInit(&mr, pHandle->meta, 0); @@ -2802,6 +2804,9 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) { SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode; pOperator = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pUser, pTaskInfo); + } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN == type) { + STableCountScanPhysiNode* pTblCountScanNode = (STableCountScanPhysiNode*)pPhyNode; + pOperator = createTableCountScanOperatorInfo(pHandle, pTblCountScanNode, pTaskInfo); } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) { STagScanPhysiNode* pScanPhyNode = (STagScanPhysiNode*)pPhyNode; diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 77d2aeba28..c42089eee5 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -4839,3 +4839,83 @@ _error: taosMemoryFree(pOperator); return NULL; } + +// ==================================================================================================================== +// TableCountScanOperator +static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator); +static void destoryTableCountScanOperator(void* param); + +SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* readHandle, STableCountScanPhysiNode* pScanNode, + SExecTaskInfo* pTaskInfo) { + int32_t code = TSDB_CODE_SUCCESS; + + STableCountScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STableCountScanInfo)); + SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); + + if (!pInfo || !pOperator) { + goto _error; + } + + pInfo->readHandle = *readHandle; + + if (pScanNode->pScanPseudoCols != NULL) { + SExprSupp* pSup = &pInfo->pseudoSup; + pSup->pExprInfo = createExprInfo(pScanNode->pScanPseudoCols, NULL, &pSup->numOfExprs); + pSup->pCtx = createSqlFunctionCtx(pSup->pExprInfo, pSup->numOfExprs, &pSup->rowEntryInfoOffset); + } + + SDataBlockDescNode* pDescNode = pScanNode->node.pOutputDataBlockDesc; + initResultSizeInfo(&pOperator->resultInfo, 1); + pInfo->pRes = createResDataBlock(pDescNode); + blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); + + tNameAssign(&pInfo->name, &pScanNode->tableName); + + pInfo->suid = pScanNode->suid; // 0 for super table, super table uid for child table + pInfo->uid = pScanNode->uid; // super table uid, or child table uid + pInfo->tableType = pScanNode->tableType; + + setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED, + pInfo, pTaskInfo); + pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator, NULL); + return pOperator; + +_error: + if (pInfo != NULL) { + destoryTableCountScanOperator(pInfo); + } + taosMemoryFreeClear(pOperator); + pTaskInfo->code = code; + return NULL; +} + +static SSDataBlock* doTableCountScan(SOperatorInfo* pOperator) { + SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; + STableCountScanInfo* pInfo = pOperator->info; + + SExprSupp* pSup = &pInfo->pseudoSup; + if (pSup->numOfExprs != 1 || pSup->pExprInfo[0].pExpr->nodeType != QUERY_NODE_FUNCTION || + pSup->pExprInfo[0].pExpr->_function.functionType != FUNCTION_TYPE_TABLE_COUNT ) { + qError("%s table count scan operator invalid pseduo columns", GET_TASKID(pTaskInfo)); + } + SColumnInfoData* pColInfoData = taosArrayGet(pInfo->pRes->pDataBlock, 0); + if (pInfo->uid != 0) { + SMetaStbStats stats = {0}; + metaGetStbStats(pInfo->readHandle.meta, pInfo->uid, &stats); + int64_t ctbNum = stats.ctbNum; + //TODO: wxy about the return type bigint or int? + colDataAppend(pColInfoData, 0, (char*)&ctbNum, false); + pInfo->pRes->info.rows = 1; + } else { + //TODO: get table count in this vnode? + } + return NULL; +} + +static void destoryTableCountScanOperator(void* param) { + STableCountScanInfo* pTableCountScanInfo = param; + blockDataDestroy(pTableCountScanInfo->pRes); + + cleanupExprSupp(&pTableCountScanInfo->pseudoSup); + taosMemoryFreeClear(param); +} -- GitLab