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

enh: optimize statement for querying the number of sub tables of the super table

This commit is contained in:
Xiaoyu Wang 2022-11-15 16:38:51 +08:00
parent d7b74c38d8
commit 40c7562470
13 changed files with 186 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
include/libs/function/functionMgt.h
View File

@ -130,6 +130,7 @@ typedef enum EFunctionType {
FUNCTION_TYPE_GROUP_KEY,
FUNCTION_TYPE_CACHE_LAST_ROW,
FUNCTION_TYPE_CACHE_LAST,
FUNCTION_TYPE_TABLE_COUNT,
// distributed splitting functions
FUNCTION_TYPE_APERCENTILE_PARTIAL = 4000,

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

@ -227,6 +227,7 @@ typedef enum ENodeType {
QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN,
QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN,
QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN,
QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN,
QUERY_NODE_PHYSICAL_PLAN_PROJECT,
QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN,
QUERY_NODE_PHYSICAL_PLAN_HASH_AGG,

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

@ -62,7 +62,8 @@ typedef enum EScanType {
SCAN_TYPE_STREAM,
SCAN_TYPE_TABLE_MERGE,
SCAN_TYPE_BLOCK_INFO,
SCAN_TYPE_LAST_ROW
SCAN_TYPE_LAST_ROW,
SCAN_TYPE_TABLE_COUNT
} EScanType;
typedef struct SScanLogicNode {
@ -314,6 +315,7 @@ typedef struct SScanPhysiNode {
typedef SScanPhysiNode STagScanPhysiNode;
typedef SScanPhysiNode SBlockDistScanPhysiNode;
typedef SScanPhysiNode STableCountScanPhysiNode;
typedef struct SLastRowScanPhysiNode {
SScanPhysiNode scan;

17
source/libs/function/src/builtins.c
View File

@ -1519,7 +1519,7 @@ static int32_t translateInterp(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
int32_t numOfParams = LIST_LENGTH(pFunc->pParameterList);
uint8_t dbPrec = pFunc->node.resType.precision;
//if (1 != numOfParams && 3 != numOfParams && 4 != numOfParams) {
// if (1 != numOfParams && 3 != numOfParams && 4 != numOfParams) {
if (1 != numOfParams) {
return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
}
@ -2060,6 +2060,11 @@ static int32_t translateTagsPseudoColumn(SFunctionNode* pFunc, char* pErrBuf, in
return TSDB_CODE_SUCCESS;
}
static int32_t translateTableCountPseudoColumn(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes, .type = TSDB_DATA_TYPE_INT};
return TSDB_CODE_SUCCESS;
}
// clang-format off
const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{
@ -3218,6 +3223,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_table_count",
.type = FUNCTION_TYPE_TABLE_COUNT,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_SCAN_PC_FUNC,
.translateFunc = translateTableCountPseudoColumn,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
};
// clang-format on

8
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -221,6 +221,8 @@ const char* nodesNodeName(ENodeType type) {
return "PhysiTableScan";
case QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN:
return "PhysiTableSeqScan";
case QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN:
return "PhysiTableMergeScan";
case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
return "PhysiSreamScan";
case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN:
@ -229,8 +231,8 @@ const char* nodesNodeName(ENodeType type) {
return "PhysiBlockDistScan";
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return "PhysiLastRowScan";
case QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN:
return "PhysiTableMergeScan";
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
return "PhysiTableCountScan";
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return "PhysiProject";
case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN:
@ -4630,6 +4632,7 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
return logicPlanToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
return physiScanNodeToJson(pObj, pJson);
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return physiLastRowScanNodeToJson(pObj, pJson);
@ -4786,6 +4789,7 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToLogicPlan(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
return jsonToPhysiScanNode(pJson, pObj);
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return jsonToPhysiLastRowScanNode(pJson, pObj);

2
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -3630,6 +3630,7 @@ static int32_t specificNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
break;
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
code = physiScanNodeToMsg(pObj, pEncoder);
break;
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
@ -3768,6 +3769,7 @@ static int32_t msgToSpecificNode(STlvDecoder* pDecoder, void* pObj) {
break;
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
code = msgToPhysiScanNode(pDecoder, pObj);
break;
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:

3
source/libs/nodes/src/nodesUtilFuncs.c
View File

@ -493,6 +493,8 @@ SNode* nodesMakeNode(ENodeType type) {
return makeNode(type, sizeof(SBlockDistScanPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN:
return makeNode(type, sizeof(SLastRowScanPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
return makeNode(type, sizeof(STableCountScanPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
return makeNode(type, sizeof(SProjectPhysiNode));
case QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN:
@ -1118,6 +1120,7 @@ void nodesDestroyNode(SNode* pNode) {
case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN:
case QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN:
destroyScanPhysiNode((SScanPhysiNode*)pNode);
break;
case QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN: {

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

@ -7464,6 +7464,112 @@ static int32_t rewriteFlushDatabase(STranslateContext* pCxt, SQuery* pQuery) {
return code;
}
static bool isTableCountProject(SNodeList* pProjectionList) {
if (1 != LIST_LENGTH(pProjectionList)) {
return false;
}
SNode* pProj = nodesListGetNode(pProjectionList, 0);
return QUERY_NODE_FUNCTION == nodeType(pProj) && 0 == strcmp(((SFunctionNode*)pProj)->functionName, "count");
}
static bool isTableCountFrom(SNode* pTable) {
if (NULL == pTable || QUERY_NODE_REAL_TABLE != nodeType(pTable)) {
return false;
}
SRealTableNode* pRtable = (SRealTableNode*)pTable;
return 0 == strcmp(pRtable->table.dbName, TSDB_INFORMATION_SCHEMA_DB) &&
0 == strcmp(pRtable->table.tableName, TSDB_INS_TABLE_TABLES);
}
static bool isTableCountCond(SNode* pCond, const char* pCol) {
if (QUERY_NODE_OPERATOR != nodeType(pCond) || OP_TYPE_EQUAL != ((SOperatorNode*)pCond)->opType) {
return false;
}
SNode* pLeft = ((SOperatorNode*)pCond)->pLeft;
SNode* pRight = ((SOperatorNode*)pCond)->pRight;
if (QUERY_NODE_COLUMN == nodeType(pLeft) && QUERY_NODE_VALUE == nodeType(pRight)) {
return 0 == strcmp(((SColumnNode*)pLeft)->colName, pCol);
}
if (QUERY_NODE_COLUMN == nodeType(pRight) && QUERY_NODE_VALUE == nodeType(pLeft)) {
return 0 == strcmp(((SColumnNode*)pRight)->colName, pCol);
}
return false;
}
static bool isTableCountWhere(SNode* pWhere) {
if (NULL == pWhere || QUERY_NODE_LOGIC_CONDITION != nodeType(pWhere)) {
return false;
}
SLogicConditionNode* pLogicCond = (SLogicConditionNode*)pWhere;
if (LOGIC_COND_TYPE_AND != pLogicCond->condType || 2 != LIST_LENGTH(pLogicCond->pParameterList)) {
return false;
}
SNode* pCond1 = nodesListGetNode(pLogicCond->pParameterList, 0);
SNode* pCond2 = nodesListGetNode(pLogicCond->pParameterList, 1);
return (isTableCountCond(pCond1, "db_name") && isTableCountCond(pCond2, "stable_name")) ||
(isTableCountCond(pCond1, "stable_name") && isTableCountCond(pCond2, "db_name"));
}
static bool isTableCountQuery(const SSelectStmt* pSelect) {
if (!isTableCountProject(pSelect->pProjectionList) || !isTableCountFrom(pSelect->pFromTable) ||
!isTableCountWhere(pSelect->pWhere) || NULL != pSelect->pPartitionByList || NULL != pSelect->pWindow ||
NULL != pSelect->pGroupByList || NULL != pSelect->pHaving || NULL != pSelect->pRange || NULL != pSelect->pEvery ||
NULL != pSelect->pFill) {
return false;
}
return true;
}
static SNode* createTableCountPseudoColumn() {
SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
if (NULL == pFunc) {
return NULL;
}
snprintf(pFunc->functionName, sizeof(pFunc->functionName), "%s", "_table_count");
return (SNode*)pFunc;
}
static int32_t rewriteCountFuncForTableCount(SSelectStmt* pSelect) {
SFunctionNode* pFunc = (SFunctionNode*)nodesListGetNode(pSelect->pProjectionList, 0);
NODES_DESTORY_LIST(pFunc->pParameterList);
snprintf(pFunc->functionName, sizeof(pFunc->functionName), "%s", "sum");
return nodesListMakeStrictAppend(&pFunc->pParameterList, createTableCountPseudoColumn());
}
static const char* getNameFromCond(SLogicConditionNode* pLogicCond, const char* pCol) {
SOperatorNode* pCond1 = (SOperatorNode*)nodesListGetNode(pLogicCond->pParameterList, 0);
SOperatorNode* pCond2 = (SOperatorNode*)nodesListGetNode(pLogicCond->pParameterList, 1);
if (QUERY_NODE_COLUMN == nodeType(pCond1->pLeft) && 0 == strcmp(((SColumnNode*)pCond1->pLeft)->colName, pCol)) {
return ((SValueNode*)pCond1->pRight)->literal;
}
if (QUERY_NODE_COLUMN == nodeType(pCond1->pRight) && 0 == strcmp(((SColumnNode*)pCond1->pRight)->colName, pCol)) {
return ((SValueNode*)pCond1->pLeft)->literal;
}
if (QUERY_NODE_COLUMN == nodeType(pCond2->pLeft) && 0 == strcmp(((SColumnNode*)pCond2->pLeft)->colName, pCol)) {
return ((SValueNode*)pCond2->pRight)->literal;
}
return ((SValueNode*)pCond2->pLeft)->literal;
}
static int32_t rewriteRealTableForTableCount(SSelectStmt* pSelect) {
STableNode* pTable = (STableNode*)pSelect->pFromTable;
snprintf(pTable->dbName, sizeof(pTable->dbName), "%s",
getNameFromCond((SLogicConditionNode*)pSelect->pWhere, "db_name"));
snprintf(pTable->tableName, sizeof(pTable->tableName), "%s",
getNameFromCond((SLogicConditionNode*)pSelect->pWhere, "stable_name"));
nodesDestroyNode(pSelect->pWhere);
pSelect->pWhere = NULL;
return TSDB_CODE_SUCCESS;
}
static int32_t rewriteTableCountQuery(SSelectStmt* pSelect) {
int32_t code = rewriteCountFuncForTableCount(pSelect);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteRealTableForTableCount(pSelect);
}
return code;
}
static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
int32_t code = TSDB_CODE_SUCCESS;
switch (nodeType(pQuery->pRoot)) {
@ -7524,6 +7630,11 @@ static int32_t rewriteQuery(STranslateContext* pCxt, SQuery* pQuery) {
case QUERY_NODE_FLUSH_DATABASE_STMT:
code = rewriteFlushDatabase(pCxt, pQuery);
break;
case QUERY_NODE_SELECT_STMT:
if (isTableCountQuery((SSelectStmt*)pQuery->pRoot)) {
code = rewriteTableCountQuery((SSelectStmt*)pQuery->pRoot);
}
break;
default:
break;
}

7
source/libs/parser/test/mockCatalog.cpp
View File

@ -65,9 +65,10 @@ void generateInformationSchema(MockCatalogService* mcs) {
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("stable_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN)
.done();
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_TABLES, TSDB_SYSTEM_TABLE, 2)
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_TABLES, TSDB_SYSTEM_TABLE, 3)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
.addColumn("table_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN)
.addColumn("stable_name", TSDB_DATA_TYPE_BINARY, TSDB_TABLE_NAME_LEN)
.done();
mcs->createTableBuilder(TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_TABLE_DISTRIBUTED, TSDB_SYSTEM_TABLE, 2)
.addColumn("db_name", TSDB_DATA_TYPE_BINARY, TSDB_DB_NAME_LEN)
@ -248,8 +249,8 @@ int32_t __catalogGetTableDistVgInfo(SCatalog* pCtg, SRequestConnInfo* pConn, con
return g_mockCatalogService->catalogGetTableDistVgInfo(pTableName, pVgList);
}
int32_t __catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId,
int32_t* tableNum, int64_t* stateTs) {
int32_t __catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId, int32_t* tableNum,
int64_t* stateTs) {
return 0;
}

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

@ -203,11 +203,13 @@ static EScanType getScanType(SLogicPlanContext* pCxt, SNodeList* pScanPseudoCols
}
if (NULL == pScanCols) {
return NULL == pScanPseudoCols
? SCAN_TYPE_TABLE
: ((FUNCTION_TYPE_BLOCK_DIST_INFO == ((SFunctionNode*)nodesListGetNode(pScanPseudoCols, 0))->funcType)
if (NULL == pScanPseudoCols) {
return SCAN_TYPE_TABLE;
}
int32_t funcType = ((SFunctionNode*)nodesListGetNode(pScanPseudoCols, 0))->funcType;
return FUNCTION_TYPE_BLOCK_DIST_INFO == funcType
? SCAN_TYPE_BLOCK_INFO
: SCAN_TYPE_TABLE);
: (FUNCTION_TYPE_TABLE_COUNT == funcType ? SCAN_TYPE_TABLE_COUNT : SCAN_TYPE_TABLE);
}
return SCAN_TYPE_TABLE;
@ -286,6 +288,8 @@ static int32_t makeScanLogicNode(SLogicPlanContext* pCxt, SRealTableNode* pRealT
return TSDB_CODE_SUCCESS;
}
static bool needScanDefaultCol(EScanType scanType) { return SCAN_TYPE_TABLE_COUNT != scanType; }
static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SRealTableNode* pRealTable,
SLogicNode** pLogicNode) {
SScanLogicNode* pScan = NULL;
@ -320,7 +324,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
pScan->hasNormalCols = true;
}
if (TSDB_CODE_SUCCESS == code) {
if (TSDB_CODE_SUCCESS == code && needScanDefaultCol(pScan->scanType)) {
code = addDefaultScanCol(pRealTable->pMeta, &pScan->pScanCols);
}

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

@ -489,6 +489,8 @@ static ENodeType getScanOperatorType(EScanType scanType) {
return QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN;
case SCAN_TYPE_BLOCK_INFO:
return QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN;
case SCAN_TYPE_TABLE_COUNT:
return QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN;
default:
break;
}
@ -623,6 +625,7 @@ static int32_t createScanPhysiNode(SPhysiPlanContext* pCxt, SSubplan* pSubplan,
switch (pScanLogicNode->scanType) {
case SCAN_TYPE_TAG:
case SCAN_TYPE_BLOCK_INFO:
case SCAN_TYPE_TABLE_COUNT:
return createSimpleScanPhysiNode(pCxt, pSubplan, pScanLogicNode, pPhyNode);
case SCAN_TYPE_LAST_ROW:
return createLastRowScanPhysiNode(pCxt, pSubplan, pScanLogicNode, pPhyNode);

17
source/libs/planner/src/planSpliter.c
View File

@ -292,6 +292,20 @@ static bool stbSplNeedSplitJoin(bool streamQuery, SJoinLogicNode* pJoin) {
return true;
}
static bool stbSplIsTableCountQuery(SLogicNode* pNode) {
if (1 != LIST_LENGTH(pNode->pChildren)) {
return false;
}
SNode* pChild = nodesListGetNode(pNode->pChildren, 0);
if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pChild)) {
if (1 != LIST_LENGTH(((SLogicNode*)pChild)->pChildren)) {
return false;
}
pChild = nodesListGetNode(((SLogicNode*)pChild)->pChildren, 0);
}
return QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pChild) && SCAN_TYPE_TABLE_COUNT == ((SScanLogicNode*)pChild)->scanType;
}
static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_SCAN:
@ -301,7 +315,8 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
case QUERY_NODE_LOGIC_PLAN_PARTITION:
return streamQuery ? false : stbSplIsMultiTbScanChild(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_AGG:
return !stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
return !stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) && stbSplHasMultiTbScan(streamQuery, pNode) &&
!stbSplIsTableCountQuery(pNode);
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return stbSplNeedSplitWindow(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_SORT:

6
source/libs/planner/test/planSysTbTest.cpp
View File

@ -38,3 +38,9 @@ TEST_F(PlanSysTableTest, withAgg) {
run("SELECT COUNT(1) FROM ins_users");
}
TEST_F(PlanSysTableTest, tableCount) {
useDb("root", "information_schema");
run("SELECT COUNT(*) FROM ins_tables WHERE db_name = 'test' AND stable_name = 'st1'");
}