diff --git a/source/libs/executor/src/sysscanoperator.c b/source/libs/executor/src/sysscanoperator.c index a88f673e0b33135f41c1fa5e868d61c66942ee9f..ac32b54f56d77795629760e4ebbc15d1bdab2d45 100644 --- a/source/libs/executor/src/sysscanoperator.c +++ b/source/libs/executor/src/sysscanoperator.c @@ -65,6 +65,8 @@ typedef struct SSysTableScanInfo { SSDataBlock* pRes; int64_t numOfBlocks; // extract basic running information. SLoadRemoteDataInfo loadInfo; + + int32_t tbnameSlotId; } SSysTableScanInfo; typedef struct { @@ -346,6 +348,11 @@ static int32_t optSysTabFilteImpl(void* arg, SNode* cond, SArray* result); static int32_t optSysCheckOper(SNode* pOpear); static int32_t optSysMergeRslt(SArray* mRslt, SArray* rslt); +static SSDataBlock* sysTableScanFromMNode(SOperatorInfo* pOperator, SSysTableScanInfo* pInfo, const char* name, + SExecTaskInfo* pTaskInfo); +void extractTbnameSlotId(SSysTableScanInfo* pInfo, const SScanPhysiNode* pScanNode); +static SSDataBlock* sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo, + const char* name, SSDataBlock* pBlock); __optSysFilter optSysGetFilterFunc(int32_t ctype, bool* reverse) { if (ctype == OP_TYPE_LOWER_EQUAL || ctype == OP_TYPE_LOWER_THAN) { *reverse = true; @@ -1309,84 +1316,112 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { getDBNameFromCondition(pInfo->pCondition, dbName); sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName); } - + SSDataBlock* pBlock = NULL; if (strncasecmp(name, TSDB_INS_TABLE_TABLES, TSDB_TABLE_FNAME_LEN) == 0) { - return sysTableScanUserTables(pOperator); + pBlock = sysTableScanUserTables(pOperator); } else if (strncasecmp(name, TSDB_INS_TABLE_TAGS, TSDB_TABLE_FNAME_LEN) == 0) { - return sysTableScanUserTags(pOperator); + pBlock = sysTableScanUserTags(pOperator); } else if (strncasecmp(name, TSDB_INS_TABLE_STABLES, TSDB_TABLE_FNAME_LEN) == 0 && pInfo->showRewrite && IS_SYS_DBNAME(dbName)) { - return sysTableScanUserSTables(pOperator); + pBlock = sysTableScanUserSTables(pOperator); } else { // load the meta from mnode of the given epset - if (pOperator->status == OP_EXEC_DONE) { - return NULL; - } - - while (1) { - int64_t startTs = taosGetTimestampUs(); - tstrncpy(pInfo->req.tb, tNameGetTableName(&pInfo->name), tListLen(pInfo->req.tb)); - tstrncpy(pInfo->req.user, pInfo->pUser, tListLen(pInfo->req.user)); + pBlock = sysTableScanFromMNode(pOperator, pInfo, name, pTaskInfo); + } - int32_t contLen = tSerializeSRetrieveTableReq(NULL, 0, &pInfo->req); - char* buf1 = taosMemoryCalloc(1, contLen); - tSerializeSRetrieveTableReq(buf1, contLen, &pInfo->req); + return sysTableScanFillTbName(pOperator, pInfo, name, pBlock); +} - // send the fetch remote task result reques - SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); - if (NULL == pMsgSendInfo) { - qError("%s prepare message %d failed", GET_TASKID(pTaskInfo), (int32_t)sizeof(SMsgSendInfo)); - pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; - return NULL; +static SSDataBlock* sysTableScanFillTbName(SOperatorInfo* pOperator, const SSysTableScanInfo* pInfo, + const char* name, SSDataBlock* pBlock) { + if (pBlock != NULL) { + if (pInfo->tbnameSlotId != -1) { + SColumnInfoData* pColumnInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, pInfo->tbnameSlotId); + char varTbName[TSDB_TABLE_FNAME_LEN - 1 + VARSTR_HEADER_SIZE] = {0}; + memcpy(varDataVal(varTbName), name, strlen(name)); + varDataSetLen(varTbName, strlen(name)); + for (int i = 0; i < pBlock->info.rows; ++i) { + colDataAppend(pColumnInfoData, i, varTbName, NULL); } + doFilterResult(pBlock, pOperator->exprSupp.pFilterInfo); + } + } + if (pBlock && pBlock->info.rows != 0) { + return pBlock; + } else { + return NULL; + } +} - int32_t msgType = (strcasecmp(name, TSDB_INS_TABLE_DNODE_VARIABLES) == 0) ? TDMT_DND_SYSTABLE_RETRIEVE - : TDMT_MND_SYSTABLE_RETRIEVE; +static SSDataBlock* sysTableScanFromMNode(SOperatorInfo* pOperator, SSysTableScanInfo* pInfo, const char* name, + SExecTaskInfo* pTaskInfo) { + if (pOperator->status == OP_EXEC_DONE) { + return NULL; + } - pMsgSendInfo->param = pOperator; - pMsgSendInfo->msgInfo.pData = buf1; - pMsgSendInfo->msgInfo.len = contLen; - pMsgSendInfo->msgType = msgType; - pMsgSendInfo->fp = loadSysTableCallback; - pMsgSendInfo->requestId = pTaskInfo->id.queryId; + while (1) { + int64_t startTs = taosGetTimestampUs(); + tstrncpy(pInfo->req.tb, tNameGetTableName(&pInfo->name), tListLen(pInfo->req.tb)); + tstrncpy(pInfo->req.user, pInfo->pUser, tListLen(pInfo->req.user)); - int64_t transporterId = 0; - int32_t code = - asyncSendMsgToServer(pInfo->readHandle.pMsgCb->clientRpc, &pInfo->epSet, &transporterId, pMsgSendInfo); - tsem_wait(&pInfo->ready); + int32_t contLen = tSerializeSRetrieveTableReq(NULL, 0, &pInfo->req); + char* buf1 = taosMemoryCalloc(1, contLen); + tSerializeSRetrieveTableReq(buf1, contLen, &pInfo->req); - if (pTaskInfo->code) { - qDebug("%s load meta data from mnode failed, totalRows:%" PRIu64 ", code:%s", GET_TASKID(pTaskInfo), - pInfo->loadInfo.totalRows, tstrerror(pTaskInfo->code)); - return NULL; - } + // send the fetch remote task result reques + SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo)); + if (NULL == pMsgSendInfo) { + qError("%s prepare message %d failed", GET_TASKID(pTaskInfo), (int32_t)sizeof(SMsgSendInfo)); + pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } - SRetrieveMetaTableRsp* pRsp = pInfo->pRsp; - pInfo->req.showId = pRsp->handle; + int32_t msgType = (strcasecmp(name, TSDB_INS_TABLE_DNODE_VARIABLES) == 0) ? TDMT_DND_SYSTABLE_RETRIEVE + : TDMT_MND_SYSTABLE_RETRIEVE; - if (pRsp->numOfRows == 0 || pRsp->completed) { - pOperator->status = OP_EXEC_DONE; - qDebug("%s load meta data from mnode completed, rowsOfSource:%d, totalRows:%" PRIu64, GET_TASKID(pTaskInfo), - pRsp->numOfRows, pInfo->loadInfo.totalRows); + pMsgSendInfo->param = pOperator; + pMsgSendInfo->msgInfo.pData = buf1; + pMsgSendInfo->msgInfo.len = contLen; + pMsgSendInfo->msgType = msgType; + pMsgSendInfo->fp = loadSysTableCallback; + pMsgSendInfo->requestId = pTaskInfo->id.queryId; - if (pRsp->numOfRows == 0) { - taosMemoryFree(pRsp); - return NULL; - } - } + int64_t transporterId = 0; + int32_t code = + asyncSendMsgToServer(pInfo->readHandle.pMsgCb->clientRpc, &pInfo->epSet, &transporterId, pMsgSendInfo); + tsem_wait(&pInfo->ready); - char* pStart = pRsp->data; - extractDataBlockFromFetchRsp(pInfo->pRes, pRsp->data, pInfo->matchInfo.pList, &pStart); - updateLoadRemoteInfo(&pInfo->loadInfo, pRsp->numOfRows, pRsp->compLen, startTs, pOperator); + if (pTaskInfo->code) { + qDebug("%s load meta data from mnode failed, totalRows:%" PRIu64 ", code:%s", GET_TASKID(pTaskInfo), + pInfo->loadInfo.totalRows, tstrerror(pTaskInfo->code)); + return NULL; + } - // todo log the filter info - doFilterResult(pInfo->pRes, pOperator->exprSupp.pFilterInfo); - taosMemoryFree(pRsp); - if (pInfo->pRes->info.rows > 0) { - return pInfo->pRes; - } else if (pOperator->status == OP_EXEC_DONE) { + SRetrieveMetaTableRsp* pRsp = pInfo->pRsp; + pInfo->req.showId = pRsp->handle; + + if (pRsp->numOfRows == 0 || pRsp->completed) { + pOperator->status = OP_EXEC_DONE; + qDebug("%s load meta data from mnode completed, rowsOfSource:%d, totalRows:%" PRIu64, GET_TASKID(pTaskInfo), + pRsp->numOfRows, pInfo->loadInfo.totalRows); + + if (pRsp->numOfRows == 0) { + taosMemoryFree(pRsp); return NULL; } } + + char* pStart = pRsp->data; + extractDataBlockFromFetchRsp(pInfo->pRes, pRsp->data, pInfo->matchInfo.pList, &pStart); + updateLoadRemoteInfo(&pInfo->loadInfo, pRsp->numOfRows, pRsp->compLen, startTs, pOperator); + + // todo log the filter info + doFilterResult(pInfo->pRes, pOperator->exprSupp.pFilterInfo); + taosMemoryFree(pRsp); + if (pInfo->pRes->info.rows > 0) { + return pInfo->pRes; + } else if (pOperator->status == OP_EXEC_DONE) { + return NULL; + } } } @@ -1407,6 +1442,8 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan goto _error; } + extractTbnameSlotId(pInfo, pScanNode); + pInfo->accountId = pScanPhyNode->accountId; pInfo->pUser = taosMemoryStrDup((void*)pUser); pInfo->sysInfo = pScanPhyNode->sysInfo; @@ -1449,6 +1486,26 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan return NULL; } +void extractTbnameSlotId(SSysTableScanInfo* pInfo, const SScanPhysiNode* pScanNode) { + pInfo->tbnameSlotId = -1; + if (pScanNode->pScanPseudoCols != NULL) { + SNode* pNode = NULL; + FOREACH(pNode, pScanNode->pScanPseudoCols) { + STargetNode* pTargetNode = NULL; + if (nodeType(pNode) == QUERY_NODE_TARGET) { + pTargetNode = (STargetNode*)pNode; + SNode* expr = pTargetNode->pExpr; + if (nodeType(expr) == QUERY_NODE_FUNCTION) { + SFunctionNode* pFuncNode = (SFunctionNode*)expr; + if (pFuncNode->funcType == FUNCTION_TYPE_TBNAME) { + pInfo->tbnameSlotId = pTargetNode->slotId; + } + } + } + } + } +} + void destroySysScanOperator(void* param) { SSysTableScanInfo* pInfo = (SSysTableScanInfo*)param; tsem_destroy(&pInfo->ready); diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index acfd5c07af0cd5c54d28ef24d586b18fbab99a20..c9fa70ff11a6909022623affd3a0ebf1850837f5 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -1230,7 +1230,6 @@ int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode * if (uvTask->rspBuf.base != NULL) { SUdfResponse rsp = {0}; void *buf = decodeUdfResponse(uvTask->rspBuf.base, &rsp); - assert(uvTask->rspBuf.len == POINTER_DISTANCE(buf, uvTask->rspBuf.base)); task->errCode = rsp.code; switch (task->type) { diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c index 6c88e4d5c8615d262db99d3e0ab5e476269820a4..e51eab453ce4f0cb688b6f6ab2eb156b10393f36 100644 --- a/source/libs/function/src/udfd.c +++ b/source/libs/function/src/udfd.c @@ -400,7 +400,6 @@ void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) { void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { SUdfdRpcSendRecvInfo *msgInfo = (SUdfdRpcSendRecvInfo *)pMsg->info.ahandle; - ASSERT(pMsg->info.ahandle != NULL); if (pEpSet) { if (!isEpsetEqual(&global.mgmtEp.epSet, pEpSet)) { diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index 612ab6375833dd901df6e0ad91d026a9da6ef7f9..03e0275d7d9a5fc7874dd1a54e4fc24973c81776 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -2388,6 +2388,9 @@ static bool tagScanOptShouldBeOptimized(SLogicNode* pNode) { if (pScan->hasNormalCols) { return false; } + if (pScan->tableType == TSDB_SYSTEM_TABLE) { + return false; + } if (NULL == pNode->pParent || QUERY_NODE_LOGIC_PLAN_AGG != nodeType(pNode->pParent) || 1 != LIST_LENGTH(pNode->pParent->pChildren)) { return false; diff --git a/source/libs/planner/src/planSpliter.c b/source/libs/planner/src/planSpliter.c index f5782dc9371fe849a69d96f9c30d6c98cec766e1..a7eac2c853659c3526a3c0527d64734a50e95115 100644 --- a/source/libs/planner/src/planSpliter.c +++ b/source/libs/planner/src/planSpliter.c @@ -333,13 +333,23 @@ static bool stbSplHasPartTbname(SNodeList* pPartKeys) { return false; } -static bool stbSplIsPartTableAgg(SAggLogicNode* pAgg) { - if (NULL != pAgg->pGroupKeys) { - return stbSplHasPartTbname(pAgg->pGroupKeys); +static bool stbSplNotSystemScan(SLogicNode* pNode) { + if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) { + return SCAN_TYPE_SYSTEM_TABLE != ((SScanLogicNode*)pNode)->scanType; + } else if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) { + return stbSplNotSystemScan((SLogicNode*)nodesListGetNode(pNode->pChildren, 0)); + } else { + return true; } +} + +static bool stbSplIsPartTableAgg(SAggLogicNode* pAgg) { if (1 != LIST_LENGTH(pAgg->node.pChildren)) { return false; } + if (NULL != pAgg->pGroupKeys) { + return stbSplHasPartTbname(pAgg->pGroupKeys) && stbSplNotSystemScan((SLogicNode*)nodesListGetNode(pAgg->node.pChildren, 0)); + } return stbSplHasPartTbname(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pAgg->node.pChildren, 0))); }