From ee34c3bb3e442786594f5c23fbc2fff17198d16b Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 21 Jul 2022 11:20:30 +0800 Subject: [PATCH] fix(query): add query involved column info --- include/libs/executor/executor.h | 2 +- source/dnode/vnode/src/inc/tq.h | 3 +- source/dnode/vnode/src/tq/tq.c | 3 +- source/dnode/vnode/src/tq/tqMeta.c | 3 +- source/libs/executor/inc/executorimpl.h | 1 + source/libs/executor/src/executor.c | 3 +- source/libs/executor/src/executorimpl.c | 46 ++++++++++++++++++------- 7 files changed, 43 insertions(+), 18 deletions(-) diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index 16e1a1c395..65244ec11a 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -64,7 +64,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers); * @param SReadHandle * @return */ -qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols); +qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchemaWrapper); /** * Set the input data block for the stream scan. diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index 07bee22a1f..cde8fd7064 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -88,7 +88,8 @@ typedef struct { STqExecTb execTb; STqExecDb execDb; }; - int32_t numOfCols; // number of out pout column, temporarily used + int32_t numOfCols; // number of out pout column, temporarily used + SSchemaWrapper *pSchemaWrapper; // columns that are involved in query } STqExecHandle; typedef struct { diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index f6862621f9..a9ed954828 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -506,7 +506,8 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { .initTqReader = true, .version = ver, }; - pHandle->execHandle.execCol.task[i] = qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, &pHandle->execHandle.numOfCols); + pHandle->execHandle.execCol.task[i] = qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, &pHandle->execHandle.numOfCols, + &pHandle->execHandle.pSchemaWrapper); ASSERT(pHandle->execHandle.execCol.task[i]); void* scanner = NULL; qExtractStreamScanner(pHandle->execHandle.execCol.task[i], &scanner); diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index 468490350a..83e852c79e 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -93,7 +93,8 @@ int32_t tqMetaOpen(STQ* pTq) { .version = handle.snapshotVer, }; - handle.execHandle.execCol.task[i] = qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols); + handle.execHandle.execCol.task[i] = qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, + &handle.execHandle.pSchemaWrapper); ASSERT(handle.execHandle.execCol.task[i]); void* scanner = NULL; qExtractStreamScanner(handle.execHandle.execCol.task[i], &scanner); diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 2cc4058b3b..3665a2539f 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -164,6 +164,7 @@ typedef struct { char* dbname; int32_t tversion; SSchemaWrapper* sw; + SSchemaWrapper* qsw; } SSchemaInfo; typedef struct SExecTaskInfo { diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 0aa50cdb0a..c92ff93cc5 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -104,7 +104,7 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO return code; } -qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols) { +qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchemaWrapper) { if (msg == NULL) { // TODO create raw scan return NULL; @@ -138,6 +138,7 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* n } } + *pSchemaWrapper = tCloneSSchemaWrapper(((SExecTaskInfo*)pTaskInfo)->schemaInfo.qsw);; return pTaskInfo; } diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 5d5eded132..2e606ae6de 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4139,32 +4139,52 @@ static STsdbReader* doCreateDataReader(STableScanPhysiNode* pTableScanNode, SRea static SArray* extractColumnInfo(SNodeList* pNodeList); -int32_t extractTableSchemaInfo(SReadHandle* pHandle, uint64_t uid, SExecTaskInfo* pTaskInfo) { +int32_t extractTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, SExecTaskInfo* pTaskInfo) { SMetaReader mr = {0}; metaReaderInit(&mr, pHandle->meta, 0); - int32_t code = metaGetTableEntryByUid(&mr, uid); + int32_t code = metaGetTableEntryByUid(&mr, pScanNode->uid); if (code != TSDB_CODE_SUCCESS) { metaReaderClear(&mr); return terrno; } - pTaskInfo->schemaInfo.tablename = strdup(mr.me.name); + SSchemaInfo* pSchemaInfo = &pTaskInfo->schemaInfo; + pSchemaInfo->tablename = strdup(mr.me.name); if (mr.me.type == TSDB_SUPER_TABLE) { - pTaskInfo->schemaInfo.sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow); - pTaskInfo->schemaInfo.tversion = mr.me.stbEntry.schemaTag.version; + pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow); + pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version; } else if (mr.me.type == TSDB_CHILD_TABLE) { tDecoderClear(&mr.coder); tb_uid_t suid = mr.me.ctbEntry.suid; metaGetTableEntryByUid(&mr, suid); - pTaskInfo->schemaInfo.sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow); - pTaskInfo->schemaInfo.tversion = mr.me.stbEntry.schemaTag.version; + pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow); + pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version; } else { - pTaskInfo->schemaInfo.sw = tCloneSSchemaWrapper(&mr.me.ntbEntry.schemaRow); + pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.ntbEntry.schemaRow); } metaReaderClear(&mr); + + int32_t numOfCols = LIST_LENGTH(pScanNode->pScanCols); + SSchemaWrapper* pqSw = taosMemoryCalloc(1, sizeof(SSchemaWrapper)); + pqSw->pSchema = taosMemoryCalloc(numOfCols, sizeof(SSchema)); + pqSw->version = pSchemaInfo->sw->version; + + for(int32_t i = 0; i < numOfCols; ++i) { + STargetNode* pNode = (STargetNode*) nodesListGetNode(pScanNode->pScanCols, i); + SColumnNode* pColNode = (SColumnNode*)pNode->pExpr; + + for(int32_t j = 0; j < pSchemaInfo->sw->nCols; ++j) { + if (pColNode->colId == pSchemaInfo->sw->pSchema[j].colId) { + pqSw->pSchema[pqSw->nCols++] = pSchemaInfo->sw->pSchema[j]; + break; + } + } + } + + pSchemaInfo->qsw = pqSw; return TSDB_CODE_SUCCESS; } @@ -4175,8 +4195,8 @@ static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) { } taosMemoryFree(pSchemaInfo->tablename); - taosMemoryFree(pSchemaInfo->sw->pSchema); - taosMemoryFree(pSchemaInfo->sw); + tDeleteSSchemaWrapper(pSchemaInfo->sw); + tDeleteSSchemaWrapper(pSchemaInfo->qsw); } static int32_t sortTableGroup(STableListInfo* pTableListInfo, int32_t groupNum) { @@ -4377,7 +4397,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo return NULL; } - code = extractTableSchemaInfo(pHandle, pTableScanNode->scan.uid, pTaskInfo); + code = extractTableSchemaInfo(pHandle, &pTableScanNode->scan, pTaskInfo); if (code) { pTaskInfo->code = terrno; return NULL; @@ -4397,7 +4417,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo return NULL; } - code = extractTableSchemaInfo(pHandle, pTableScanNode->scan.uid, pTaskInfo); + code = extractTableSchemaInfo(pHandle, &pTableScanNode->scan, pTaskInfo); if (code) { pTaskInfo->code = terrno; return NULL; @@ -4479,7 +4499,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo return NULL; } - code = extractTableSchemaInfo(pHandle, pScanNode->scan.uid, pTaskInfo); + code = extractTableSchemaInfo(pHandle, &pScanNode->scan, pTaskInfo); if (code != TSDB_CODE_SUCCESS) { pTaskInfo->code = code; return NULL; -- GitLab