xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

sort scan cols

This commit is contained in:
Xiaoyu Wang 2022-03-26 01:39:10 -04:00
parent 75d77eb126
commit f6bcae1913
11 changed files with 56 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/nodes/plannodes.h
View File

@ -65,6 +65,7 @@ typedef struct SAggLogicNode {
typedef struct SProjectLogicNode {
SLogicNode node;
SNodeList* pProjections;
char stmtName[TSDB_TABLE_NAME_LEN];
} SProjectLogicNode;
typedef struct SVnodeModifLogicNode {

1
include/libs/nodes/querynodes.h
View File

@ -231,6 +231,7 @@ typedef struct SSelectStmt {
SNodeList* pOrderByList; // SOrderByExprNode
SNode* pLimit;
SNode* pSlimit;
char stmtName[TSDB_TABLE_NAME_LEN];
} SSelectStmt;
typedef enum ESetOperatorType {

13
source/libs/executor/src/executorimpl.c
View File

@ -5557,7 +5557,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo *pOperator, bool* newgroup) {
blockDataCleanup(pInfo->pRes);
int32_t tableNameSlotId = 0;
int32_t tableNameSlotId = 1;
SColumnInfoData* pTableNameCol = taosArrayGet(pInfo->pRes->pDataBlock, tableNameSlotId);
char * name = NULL;
@ -8644,9 +8644,14 @@ SArray* extractScanColumnId(SNodeList* pNodeList) {
}
for(int32_t i = 0; i < numOfCols; ++i) {
STargetNode* pNode = (STargetNode*) nodesListGetNode(pNodeList, i);
SColumnNode* pColNode = (SColumnNode*) pNode->pExpr;
taosArrayPush(pList, &pColNode->colId);
for (int32_t j = 0; j < numOfCols; ++j) {
STargetNode* pNode = (STargetNode*) nodesListGetNode(pNodeList, j);
if (pNode->slotId == i) {
SColumnNode* pColNode = (SColumnNode*) pNode->pExpr;
taosArrayPush(pList, &pColNode->colId);
break;
}
}
}
return pList;

1
source/libs/nodes/src/nodesCloneFuncs.c
View File

@ -250,6 +250,7 @@ static SNode* logicAggCopy(const SAggLogicNode* pSrc, SAggLogicNode* pDst) {
static SNode* logicProjectCopy(const SProjectLogicNode* pSrc, SProjectLogicNode* pDst) {
COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
CLONE_NODE_LIST_FIELD(pProjections);
COPY_CHAR_ARRAY_FIELD(stmtName);
return (SNode*)pDst;
}

6
source/libs/parser/src/parAstCreater.c
View File

@ -645,6 +645,11 @@ SNode* createTempTableNode(SAstCreateContext* pCxt, SNode* pSubquery, const STok
tempTable->pSubquery = pSubquery;
if (NULL != pTableAlias && TK_NK_NIL != pTableAlias->type) {
strncpy(tempTable->table.tableAlias, pTableAlias->z, pTableAlias->n);
} else {
sprintf(tempTable->table.tableAlias, "%p", tempTable);
}
if (QUERY_NODE_SELECT_STMT == nodeType(pSubquery)) {
strcpy(((SSelectStmt*)pSubquery)->stmtName, tempTable->table.tableAlias);
}
return (SNode*)tempTable;
}
@ -792,6 +797,7 @@ SNode* createSelectStmt(SAstCreateContext* pCxt, bool isDistinct, SNodeList* pPr
select->isDistinct = isDistinct;
select->pProjectionList = pProjectionList;
select->pFromTable = pTable;
sprintf(select->stmtName, "%p", select);
return (SNode*)select;
}

2
source/libs/parser/src/parTranslater.c
View File

@ -1850,6 +1850,7 @@ static int32_t createSelectStmtForShow(ENodeType showType, SSelectStmt** pStmt)
if (NULL == pSelect) {
return TSDB_CODE_OUT_OF_MEMORY;
}
sprintf(pSelect->stmtName, "%p", pSelect);
SRealTableNode* pTable = nodesMakeNode(QUERY_NODE_REAL_TABLE);
if (NULL == pTable) {
@ -1858,6 +1859,7 @@ static int32_t createSelectStmtForShow(ENodeType showType, SSelectStmt** pStmt)
}
strcpy(pTable->table.dbName, TSDB_INFORMATION_SCHEMA_DB);
strcpy(pTable->table.tableName, getSysTableName(showType));
strcpy(pTable->table.tableAlias, pTable->table.tableName);
pSelect->pFromTable = (SNode*)pTable;
*pStmt = pSelect;

10
source/libs/planner/src/planLogicCreater.c
View File

@ -290,13 +290,14 @@ static int32_t createLogicNodeByTable(SLogicPlanContext* pCxt, SSelectStmt* pSel
return code;
}
static SColumnNode* createColumnByExpr(SExprNode* pExpr) {
static SColumnNode* createColumnByExpr(const char* pStmtName, SExprNode* pExpr) {
SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {
return NULL;
}
pCol->node.resType = pExpr->resType;
strcpy(pCol->colName, pExpr->aliasName);
strcpy(pCol->tableAlias, pStmtName);
return pCol;
}
@ -484,7 +485,7 @@ static int32_t createWindowLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSele
return TSDB_CODE_FAILED;
}
static int32_t createColumnByProjections(SLogicPlanContext* pCxt, SNodeList* pExprs, SNodeList** pCols) {
static int32_t createColumnByProjections(SLogicPlanContext* pCxt, const char* pStmtName, SNodeList* pExprs, SNodeList** pCols) {
SNodeList* pList = nodesMakeList();
if (NULL == pList) {
return TSDB_CODE_OUT_OF_MEMORY;
@ -492,7 +493,7 @@ static int32_t createColumnByProjections(SLogicPlanContext* pCxt, SNodeList* pEx
SNode* pNode;
FOREACH(pNode, pExprs) {
if (TSDB_CODE_SUCCESS != nodesListAppend(pList, createColumnByExpr((SExprNode*)pNode))) {
if (TSDB_CODE_SUCCESS != nodesListAppend(pList, createColumnByExpr(pStmtName, (SExprNode*)pNode))) {
nodesDestroyList(pList);
return TSDB_CODE_OUT_OF_MEMORY;
}
@ -514,9 +515,10 @@ static int32_t createProjectLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSel
if (NULL == pProject->pProjections) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
strcpy(pProject->stmtName, pSelect->stmtName);
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByProjections(pCxt,pSelect->pProjectionList, &pProject->node.pTargets);
code = createColumnByProjections(pCxt, pSelect->stmtName, pSelect->pProjectionList, &pProject->node.pTargets);
}
if (TSDB_CODE_SUCCESS == code) {

26
source/libs/planner/src/planPhysiCreater.c
View File

@ -30,14 +30,20 @@ typedef struct SPhysiPlanContext {
SArray* pExecNodeList;
} SPhysiPlanContext;
static int32_t getSlotKey(SNode* pNode, char* pKey) {
static int32_t getSlotKey(SNode* pNode, const char* pStmtName, char* pKey) {
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
if (NULL != pStmtName) {
return sprintf(pKey, "%s.%s", pStmtName, pCol->node.aliasName);
}
if ('\0' == pCol->tableAlias[0]) {
return sprintf(pKey, "%s", pCol->colName);
}
return sprintf(pKey, "%s.%s", pCol->tableAlias, pCol->colName);
}
if (NULL != pStmtName) {
return sprintf(pKey, "%s.%s", pStmtName, ((SExprNode*)pNode)->aliasName);
}
return sprintf(pKey, "%s", ((SExprNode*)pNode)->aliasName);
}
@ -74,7 +80,7 @@ static int32_t putSlotToHashImpl(int16_t dataBlockId, int16_t slotId, const char
static int32_t putSlotToHash(int16_t dataBlockId, int16_t slotId, SNode* pNode, SHashObj* pHash) {
char name[TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN];
int32_t len = getSlotKey(pNode, name);
int32_t len = getSlotKey(pNode, NULL, name);
return putSlotToHashImpl(dataBlockId, slotId, name, len, pHash);
}
@ -138,14 +144,14 @@ static int32_t createDataBlockDesc(SPhysiPlanContext* pCxt, SNodeList* pList, SD
return code;
}
static int32_t addDataBlockSlotsImpl(SPhysiPlanContext* pCxt, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc, bool output) {
static int32_t addDataBlockSlotsImpl(SPhysiPlanContext* pCxt, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc, const char* pStmtName, bool output) {
int32_t code = TSDB_CODE_SUCCESS;
SHashObj* pHash = taosArrayGetP(pCxt->pLocationHelper, pDataBlockDesc->dataBlockId);
int16_t nextSlotId = taosHashGetSize(pHash), slotId = 0;
SNode* pNode = NULL;
FOREACH(pNode, pList) {
char name[TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN] = {0};
int32_t len = getSlotKey(pNode, name);
int32_t len = getSlotKey(pNode, pStmtName, name);
SSlotIndex* pIndex = taosHashGet(pHash, name, len);
if (NULL == pIndex) {
code = nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, pNode, nextSlotId, output));
@ -175,11 +181,15 @@ static int32_t addDataBlockSlotsImpl(SPhysiPlanContext* pCxt, SNodeList* pList,
}
static int32_t addDataBlockSlots(SPhysiPlanContext* pCxt, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc) {
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, false);
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, NULL, false);
}
static int32_t addDataBlockSlotsForProject(SPhysiPlanContext* pCxt, const char* pStmtName, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc) {
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, pStmtName, true);
}
static int32_t pushdownDataBlockSlots(SPhysiPlanContext* pCxt, SNodeList* pList, SDataBlockDescNode* pDataBlockDesc) {
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, false);
return addDataBlockSlotsImpl(pCxt, pList, pDataBlockDesc, NULL, true);
}
typedef struct SSetSlotIdCxt {
@ -192,7 +202,7 @@ static EDealRes doSetSlotId(SNode* pNode, void* pContext) {
if (QUERY_NODE_COLUMN == nodeType(pNode) && 0 != strcmp(((SColumnNode*)pNode)->colName, "*")) {
SSetSlotIdCxt* pCxt = (SSetSlotIdCxt*)pContext;
char name[TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN];
int32_t len = getSlotKey(pNode, name);
int32_t len = getSlotKey(pNode, NULL, name);
SSlotIndex* pIndex = taosHashGet(pCxt->pLeftHash, name, len);
if (NULL == pIndex) {
pIndex = taosHashGet(pCxt->pRightHash, name, len);
@ -670,7 +680,7 @@ static int32_t createProjectPhysiNode(SPhysiPlanContext* pCxt, SNodeList* pChild
int32_t code = setListSlotId(pCxt, ((SPhysiNode*)nodesListGetNode(pChildren, 0))->pOutputDataBlockDesc->dataBlockId, -1, pProjectLogicNode->pProjections, &pProject->pProjections);
if (TSDB_CODE_SUCCESS == code) {
code = addDataBlockSlots(pCxt, pProject->pProjections, pProject->node.pOutputDataBlockDesc);
code = addDataBlockSlotsForProject(pCxt, pProjectLogicNode->stmtName, pProject->pProjections, pProject->node.pOutputDataBlockDesc);
}
if (TSDB_CODE_SUCCESS == code) {
code = setConditionsSlotId(pCxt, (const SLogicNode*)pProjectLogicNode, (SPhysiNode*)pProject);

10
source/libs/planner/test/plannerTest.cpp
View File

@ -170,7 +170,7 @@ TEST_F(PlannerTest, groupBy) {
bind("SELECT count(*) FROM t1");
ASSERT_TRUE(run());
bind("SELECT c1, count(*) FROM t1 GROUP BY c1");
bind("SELECT c1, max(c3), min(c2), count(*) FROM t1 GROUP BY c1");
ASSERT_TRUE(run());
bind("SELECT c1 + c3, c1 + count(*) FROM t1 where c2 = 'abc' GROUP BY c1, c3");
@ -205,6 +205,14 @@ TEST_F(PlannerTest, showTables) {
setDatabase("root", "test");
bind("show tables");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, showStables) {
setDatabase("root", "test");
bind("show stables");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, createTopic) {

1
source/libs/qworker/inc/qworkerInt.h
View File

@ -178,6 +178,7 @@ typedef struct SQWorkerMgmt {
#define QW_TASK_ELOG(param, ...) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_WLOG(param, ...) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOG(param, ...) qDebug("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_DLOGL(param, ...) qDebugL("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId, __VA_ARGS__)
#define QW_TASK_ELOG_E(param) qError("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)
#define QW_TASK_WLOG_E(param) qWarn("QW:%p QID:0x%"PRIx64",TID:0x%"PRIx64" " param, mgmt, qId, tId)

2
source/libs/qworker/src/qworker.c
View File

@ -910,6 +910,8 @@ int32_t qwProcessQuery(QW_FPARAMS_DEF, SQWMsg *qwMsg, int8_t taskType) {
atomic_store_8(&ctx->taskType, taskType);
QW_TASK_DLOGL("task string : %s", qwMsg->msg);
code = qStringToSubplan(qwMsg->msg, &plan);
if (TSDB_CODE_SUCCESS != code) {
QW_TASK_ELOG("task string to subplan failed, code:%x - %s", code, tstrerror(code));