diff --git a/source/libs/command/inc/commandInt.h b/source/libs/command/inc/commandInt.h index 996891c77a1..f74c61ea78b 100644 --- a/source/libs/command/inc/commandInt.h +++ b/source/libs/command/inc/commandInt.h @@ -104,6 +104,7 @@ extern "C" { #define EXPLAIN_VGROUP_SLOT_FORMAT "vgroup_slot=%d,%d" #define EXPLAIN_UID_SLOT_FORMAT "uid_slot=%d,%d" #define EXPLAIN_SRC_SCAN_FORMAT "src_scan=%d,%d" +#define EXPLAIN_PLAN_BLOCKING "blocking=%d" #define COMMAND_RESET_LOG "resetLog" #define COMMAND_SCHEDULE_POLICY "schedulePolicy" diff --git a/source/libs/command/src/explain.c b/source/libs/command/src/explain.c index 655775e2242..27cfaab3cf2 100644 --- a/source/libs/command/src/explain.c +++ b/source/libs/command/src/explain.c @@ -629,6 +629,8 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pAggNode->node.pOutputDataBlockDesc->outputRowSize); EXPLAIN_ROW_APPEND_LIMIT(pAggNode->node.pLimit); EXPLAIN_ROW_APPEND_SLIMIT(pAggNode->node.pSlimit); + EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT); + EXPLAIN_ROW_APPEND(EXPLAIN_PLAN_BLOCKING, !pAggNode->node.forceCreateNonBlockingOptr); EXPLAIN_ROW_END(); QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1)); diff --git a/source/libs/planner/src/planOptimizer.c b/source/libs/planner/src/planOptimizer.c index d7983531404..28e07fe98df 100644 --- a/source/libs/planner/src/planOptimizer.c +++ b/source/libs/planner/src/planOptimizer.c @@ -2843,10 +2843,26 @@ static bool pushDownLimitTo(SLogicNode* pNodeWithLimit, SLogicNode* pNodeLimitPu case QUERY_NODE_LOGIC_PLAN_AGG: { if (nodeType(pNodeWithLimit) == QUERY_NODE_LOGIC_PLAN_PROJECT && (isPartTagAgg((SAggLogicNode*)pNodeLimitPushTo) || isPartTableAgg((SAggLogicNode*)pNodeLimitPushTo))) { - // when part by tag, slimit will be cloned to agg, and it will be pipelined. + // when part by tag/tbname, slimit will be cloned to agg, and it will be pipelined. // The scan below will do scanning with group order return cloneLimit(pNodeWithLimit, pNodeLimitPushTo, CLONE_SLIMIT); } + // else if not part by tag and tbname, the partition node below indicates that results are sorted, the agg node can + // be pipelined. + if (nodeType(pNodeWithLimit) == QUERY_NODE_LOGIC_PLAN_PROJECT && LIST_LENGTH(pNodeLimitPushTo->pChildren) == 1) { + SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pNodeLimitPushTo->pChildren, 0); + if (nodeType(pChild) == QUERY_NODE_LOGIC_PLAN_PARTITION) { + pNodeLimitPushTo->forceCreateNonBlockingOptr = true; + return cloneLimit(pNodeWithLimit, pNodeLimitPushTo, CLONE_SLIMIT); + } + // Currently, partColOpt is executed after pushDownLimitOpt, and partColOpt will replace partition node with + // sort node. + // To avoid dependencies between these two optimizations, we add sort node too. + if (nodeType(pChild) == QUERY_NODE_LOGIC_PLAN_SORT && ((SSortLogicNode*)pChild)->calcGroupId) { + pNodeLimitPushTo->forceCreateNonBlockingOptr = true; + return cloneLimit(pNodeWithLimit, pNodeLimitPushTo, CLONE_SLIMIT); + } + } break; } case QUERY_NODE_LOGIC_PLAN_SCAN: @@ -3593,6 +3609,7 @@ static SSortLogicNode* partColOptCreateSort(SPartitionLogicNode* pPartition) { nodesListMakeAppend(&pSort->pSortKeys, (SNode*)pOrder); pOrder->order = ORDER_ASC; pOrder->pExpr = nodesCloneNode(node); + pOrder->nullOrder = NULL_ORDER_FIRST; if (!pOrder->pExpr) code = TSDB_CODE_OUT_OF_MEMORY; } } diff --git a/source/libs/planner/src/planSpliter.c b/source/libs/planner/src/planSpliter.c index 881cc02062d..33d6a329eeb 100644 --- a/source/libs/planner/src/planSpliter.c +++ b/source/libs/planner/src/planSpliter.c @@ -41,6 +41,8 @@ typedef struct SSplitRule { typedef bool (*FSplFindSplitNode)(SSplitContext* pCxt, SLogicSubplan* pSubplan, SLogicNode* pNode, void* pInfo); +static int32_t stbSplCreateMergeKeys(SNodeList* pSortKeys, SNodeList* pTargets, SNodeList** pOutput); + static void splSetSubplanVgroups(SLogicSubplan* pSubplan, SLogicNode* pNode) { if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) { TSWAP(pSubplan->pVgroupList, ((SScanLogicNode*)pNode)->pVgroupList); @@ -883,15 +885,119 @@ static int32_t stbSplSplitAggNodeForPartTable(SSplitContext* pCxt, SStableSplitI return code; } +static SFunctionNode* createGroupKeyAggFunc(SColumnNode* pGroupCol) { + SFunctionNode* pFunc = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION); + if (pFunc) { + strcpy(pFunc->functionName, "_group_key"); + strcpy(pFunc->node.aliasName, pGroupCol->node.aliasName); + strcpy(pFunc->node.userAlias, pGroupCol->node.userAlias); + int32_t code = nodesListMakeStrictAppend(&pFunc->pParameterList, nodesCloneNode((SNode*)pGroupCol)); + if (code == TSDB_CODE_SUCCESS) { + code = fmGetFuncInfo(pFunc, NULL, 0); + } + if (TSDB_CODE_SUCCESS != code) { + nodesDestroyNode((SNode*)pFunc); + pFunc = NULL; + } + char name[TSDB_FUNC_NAME_LEN + TSDB_NAME_DELIMITER_LEN + TSDB_POINTER_PRINT_BYTES + 1] = {0}; + int32_t len = snprintf(name, sizeof(name) - 1, "%s.%p", pFunc->functionName, pFunc); + taosCreateMD5Hash(name, len); + strncpy(pFunc->node.aliasName, name, TSDB_COL_NAME_LEN - 1); + } + return pFunc; +} + +/** + * @brief For pipelined agg node, add a SortMergeNode to merge result from vnodes. + * For agg + partition, results are sorted by group id, use group sort. + * For agg + sort for group, results are sorted by partition keys, not group id, merges keys should be the same + * as partition keys + */ +static int32_t stbSplAggNodeCreateMerge(SSplitContext* pCtx, SStableSplitInfo* pInfo, SLogicNode* pChildAgg) { + bool groupSort = true; + SNodeList* pMergeKeys = NULL; + int32_t code = TSDB_CODE_SUCCESS; + bool sortForGroup = false; + + if (pChildAgg->pChildren->length != 1) return TSDB_CODE_TSC_INTERNAL_ERROR; + + SLogicNode* pChild = (SLogicNode*)nodesListGetNode(pChildAgg->pChildren, 0); + if (nodeType(pChild) == QUERY_NODE_LOGIC_PLAN_SORT) { + SSortLogicNode* pSort = (SSortLogicNode*)pChild; + if (pSort->calcGroupId) { + SNode *node, *node2; + groupSort = false; + sortForGroup = true; + SNodeList* extraAggFuncs = NULL; + uint32_t originalLen = LIST_LENGTH(pSort->node.pTargets), idx = 0; + code = stbSplCreateMergeKeys(pSort->pSortKeys, pSort->node.pTargets, &pMergeKeys); + if (TSDB_CODE_SUCCESS != code) return code; + + // Create group_key func for all sort keys. + // We only need newly added nodes in pSort.node.pTargets when stbSplCreateMergeKeys + FOREACH(node, pSort->node.pTargets) { + if (idx++ < originalLen) continue; + SFunctionNode* pGroupKeyFunc = createGroupKeyAggFunc((SColumnNode*)node); + if (!pGroupKeyFunc) { + code = TSDB_CODE_OUT_OF_MEMORY; + break; + } + code = nodesListMakeStrictAppend(&extraAggFuncs, (SNode*)pGroupKeyFunc); + if (code != TSDB_CODE_SUCCESS) { + nodesDestroyNode((SNode*)pGroupKeyFunc); + } + } + + if (TSDB_CODE_SUCCESS == code) { + // add these extra group_key funcs into targets + code = createColumnByRewriteExprs(extraAggFuncs, &pChildAgg->pTargets); + } + if (code == TSDB_CODE_SUCCESS) { + nodesListAppendList(((SAggLogicNode*)pChildAgg)->pAggFuncs, extraAggFuncs); + extraAggFuncs = NULL; + } + + if (code == TSDB_CODE_SUCCESS) { + FOREACH(node, pMergeKeys) { + SOrderByExprNode* pOrder = (SOrderByExprNode*)node; + SColumnNode* pCol = (SColumnNode*)pOrder->pExpr; + FOREACH(node2, ((SAggLogicNode*)pChildAgg)->pAggFuncs) { + SFunctionNode* pFunc = (SFunctionNode*)node2; + if (0 != strcmp(pFunc->functionName, "_group_key")) continue; + SNode* pParam = nodesListGetNode(pFunc->pParameterList, 0); + if (!nodesEqualNode(pParam, (SNode*)pCol)) continue; + + // use the colName of group_key func to make sure finding the right slot id for merge keys. + strcpy(pCol->colName, pFunc->node.aliasName); + strcpy(pCol->node.aliasName, pFunc->node.aliasName); + memset(pCol->tableAlias, 0, TSDB_TABLE_NAME_LEN); + break; + } + } + } + if (TSDB_CODE_SUCCESS != code) { + nodesDestroyList(pMergeKeys); + nodesDestroyList(extraAggFuncs); + } + } + } + code = stbSplCreateMergeNode(pCtx, NULL, pInfo->pSplitNode, pMergeKeys, pChildAgg, groupSort); + if (TSDB_CODE_SUCCESS == code && sortForGroup) { + SMergeLogicNode* pMerge = + (SMergeLogicNode*)nodesListGetNode(pInfo->pSplitNode->pChildren, LIST_LENGTH(pInfo->pSplitNode->pChildren) - 1); + pMerge->inputWithGroupId = true; + } + return code; +} + static int32_t stbSplSplitAggNodeForCrossTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) { SLogicNode* pPartAgg = NULL; int32_t code = stbSplCreatePartAggNode((SAggLogicNode*)pInfo->pSplitNode, &pPartAgg); - if (TSDB_CODE_SUCCESS == code) { // if slimit was pushed down to agg, agg will be pipelined mode, add sort merge before parent agg - if ((SAggLogicNode*)pInfo->pSplitNode->pSlimit) - code = stbSplCreateMergeNode(pCxt, NULL, pInfo->pSplitNode, NULL, pPartAgg, true); + if (pInfo->pSplitNode->forceCreateNonBlockingOptr) + code = stbSplAggNodeCreateMerge(pCxt, pInfo, pPartAgg); else code = stbSplCreateExchangeNode(pCxt, pInfo->pSplitNode, pPartAgg); } else { diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 9c6cd5c99af..18369ed23b9 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -53,6 +53,10 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col.py -Q 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col.py -Q 2 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col.py +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col_agg.py +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col_agg.py -Q 2 +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col_agg.py -Q 3 +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_by_col_agg.py -Q 4 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqShow.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeStb0.py diff --git a/tests/system-test/2-query/partition_by_col_agg.py b/tests/system-test/2-query/partition_by_col_agg.py new file mode 100644 index 00000000000..c522eb1e2bb --- /dev/null +++ b/tests/system-test/2-query/partition_by_col_agg.py @@ -0,0 +1,252 @@ +import taos +import sys +import time +import socket +import os +import threading +import math +from datetime import datetime + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +# from tmqCommon import * + +COMPARE_DATA = 0 +COMPARE_LEN = 1 + +class TDTestCase: + def __init__(self): + self.vgroups = 4 + self.ctbNum = 10 + self.rowsPerTbl = 10000 + self.duraion = '1h' + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor(), False) + + def create_database(self,tsql, dbName,dropFlag=1,vgroups=2,replica=1, duration:str='1d'): + if dropFlag == 1: + tsql.execute("drop database if exists %s"%(dbName)) + + tsql.execute("create database if not exists %s vgroups %d replica %d duration %s"%(dbName, vgroups, replica, duration)) + tdLog.debug("complete to create database %s"%(dbName)) + return + + def create_stable(self,tsql, paraDict): + colString = tdCom.gen_column_type_str(colname_prefix=paraDict["colPrefix"], column_elm_list=paraDict["colSchema"]) + tagString = tdCom.gen_tag_type_str(tagname_prefix=paraDict["tagPrefix"], tag_elm_list=paraDict["tagSchema"]) + sqlString = f"create table if not exists %s.%s (%s) tags (%s)"%(paraDict["dbName"], paraDict["stbName"], colString, tagString) + tdLog.debug("%s"%(sqlString)) + tsql.execute(sqlString) + return + + def create_ctable(self,tsql=None, dbName='dbx',stbName='stb',ctbPrefix='ctb',ctbNum=1,ctbStartIdx=0): + for i in range(ctbNum): + sqlString = "create table %s.%s%d using %s.%s tags(%d, 'tb%d', 'tb%d', %d, %d, %d)" % \ + (dbName,ctbPrefix,i+ctbStartIdx,dbName,stbName,(i+ctbStartIdx) % 5,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx,i+ctbStartIdx) + tsql.execute(sqlString) + + tdLog.debug("complete to create %d child tables by %s.%s" %(ctbNum, dbName, stbName)) + return + + def insert_data(self,tsql,dbName,ctbPrefix,ctbNum,rowsPerTbl,batchNum,startTs,tsStep): + tdLog.debug("start to insert data ............") + tsql.execute("use %s" %dbName) + pre_insert = "insert into " + sql = pre_insert + + for i in range(ctbNum): + rowsBatched = 0 + sql += " %s%d values "%(ctbPrefix,i) + for j in range(rowsPerTbl): + if (i < ctbNum/2): + sql += "(%d, %d, %d, %d,%d,%d,%d,true,'binary%d', 'nchar%d') "%(startTs + j*tsStep, j%1000, j%500, j%1000, j%5000, j%5400, j%128, j%10000, j%1000) + else: + sql += "(%d, %d, NULL, %d,NULL,%d,%d,true,'binary%d', 'nchar%d') "%(startTs + j*tsStep, j%1000, j%500, j%1000, j%128, j%10000, j%1000) + rowsBatched += 1 + if ((rowsBatched == batchNum) or (j == rowsPerTbl - 1)): + tsql.execute(sql) + rowsBatched = 0 + if j < rowsPerTbl - 1: + sql = "insert into %s%d values " %(ctbPrefix,i) + else: + sql = "insert into " + if sql != pre_insert: + tsql.execute(sql) + tdLog.debug("insert data ............ [OK]") + return + + def prepareTestEnv(self): + tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") + paraDict = {'dbName': 'test', + 'dropFlag': 1, + 'vgroups': 2, + 'stbName': 'meters', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1}, + {'type': 'BIGINT', 'count':1}, + {'type': 'FLOAT', 'count':1}, + {'type': 'DOUBLE', 'count':1}, + {'type': 'smallint', 'count':1}, + {'type': 'tinyint', 'count':1}, + {'type': 'bool', 'count':1}, + {'type': 'binary', 'len':10, 'count':1}, + {'type': 'nchar', 'len':10, 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'nchar', 'len':20, 'count':1},{'type': 'binary', 'len':20, 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'smallint', 'count':1},{'type': 'DOUBLE', 'count':1}], + 'ctbPrefix': 't', + 'ctbStartIdx': 0, + 'ctbNum': 100, + 'rowsPerTbl': 10000, + 'batchNum': 3000, + 'startTs': 1537146000000, + 'tsStep': 600000} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + tdLog.info("create database") + self.create_database(tsql=tdSql, dbName=paraDict["dbName"], dropFlag=paraDict["dropFlag"], vgroups=paraDict["vgroups"], replica=self.replicaVar, duration=self.duraion) + + tdLog.info("create stb") + self.create_stable(tsql=tdSql, paraDict=paraDict) + + tdLog.info("create child tables") + self.create_ctable(tsql=tdSql, dbName=paraDict["dbName"], \ + stbName=paraDict["stbName"],ctbPrefix=paraDict["ctbPrefix"],\ + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict["ctbStartIdx"]) + self.insert_data(tsql=tdSql, dbName=paraDict["dbName"],\ + ctbPrefix=paraDict["ctbPrefix"],ctbNum=paraDict["ctbNum"],\ + rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"],\ + startTs=paraDict["startTs"],tsStep=paraDict["tsStep"]) + return + + def check_explain_res_has_row(self, plan_str_expect: str, rows): + plan_found = False + for row in rows: + if str(row).find(plan_str_expect) >= 0: + tdLog.debug("plan: [%s] found in: [%s]" % (plan_str_expect, str(row))) + plan_found = True + break + if not plan_found: + tdLog.exit("plan: %s not found in res: [%s]" % (plan_str_expect, str(rows))) + + + def test_sort_for_partition_hint(self): + pass + + def add_order_by(self, sql: str, order_by: str, select_list: str = "*") -> str: + return "select %s from (%s)t order by %s" % (select_list, sql, order_by) + + def add_hint(self, sql: str) -> str: + return "select /*+ sort_for_group() */ %s" % sql[6:] + + def query_with_time(self, sql): + start = datetime.now() + tdSql.query(sql, queryTimes=1) + return (datetime.now().timestamp() - start.timestamp()) * 1000 + + def explain_sql(self, sql: str): + sql = "explain verbose true " + sql + tdSql.query(sql, queryTimes=1) + return tdSql.queryResult + + def query_and_compare_res(self, sql1, sql2, compare_what: int = 0): + dur = self.query_with_time(sql1) + tdLog.debug("sql1 query with time: [%f]" % dur) + res1 = tdSql.queryResult + dur = self.query_with_time(sql2) + tdLog.debug("sql2 query with time: [%f]" % dur) + res2 = tdSql.queryResult + if res1 is None or res2 is None: + tdLog.exit("res1 or res2 is None") + if compare_what <= COMPARE_LEN: + if len(res1) != len(res2): + tdLog.exit("query and copare failed cause different rows, sql1: [%s], rows: [%d], sql2: [%s], rows: [%d]" % (sql1, len(res1), sql2, len(res2))) + if compare_what == COMPARE_DATA: + for i in range(0, len(res1)): + if res1[i] != res2[i]: + tdLog.exit("compare failed for row: [%d], sqls: [%s] res1: [%s], sql2 : [%s] res2: [%s]" % (i, sql1, res1[i], sql2, res2[i])) + tdLog.debug("sql: [%s] and sql: [%s] have same results, rows: [%d]" % (sql1, sql2, len(res1))) + + def query_and_compare_first_rows(self, sql1, sql2): + dur = self.query_with_time(sql1) + tdLog.debug("sql1 query with time: [%f]" % dur) + res1 = tdSql.queryResult + dur = self.query_with_time(sql2) + tdLog.debug("sql2 query with time: [%f]" % dur) + res2 = tdSql.queryResult + if res1 is None or res2 is None: + tdLog.exit("res1 or res2 is None") + for i in range(0, min(len(res1), len(res2))): + if res1[i] != res2[i]: + tdLog.exit("compare failed for row: [%d], sqls: [%s] res1: [%s], sql2 : [%s] res2: [%s]" % (i, sql1, res1[i], sql2, res2[i])) + tdLog.debug("sql: [%s] and sql: [%s] have same results, rows: [%d]" % (sql1, sql2, min(len(res1), len(res2)))) + + def prepare_and_query_and_compare(self, sqls: [], order_by: str, select_list: str = "*", compare_what: int = 0): + for sql in sqls: + sql_hint = self.add_order_by(self.add_hint(sql), order_by, select_list) + sql = self.add_order_by(sql, order_by, select_list) + self.check_explain_res_has_row("Sort", self.explain_sql(sql_hint)) + self.check_explain_res_has_row("Partition", self.explain_sql(sql)) + self.query_and_compare_res(sql, sql_hint, compare_what=compare_what) + + def check_explain(self, sql): + sql_hint = self.add_hint(sql) + explain_res = self.explain_sql(sql) + self.check_explain_res_has_row('SortMerge', explain_res) + self.check_explain_res_has_row("blocking=0", explain_res) + explain_res = self.explain_sql(sql_hint) + self.check_explain_res_has_row('SortMerge', explain_res) + self.check_explain_res_has_row('blocking=0', explain_res) + + def test_pipelined_agg_plan_with_slimit(self): + sql = 'select count(*), %s from meters partition by %s slimit 1' + self.check_explain(sql % ('c1','c1')) + self.check_explain(sql % ('c1,c2', 'c1,c2')) + + # should fail + # self.check_explain(sql % ('t1', 'c1,t1')) + # self.check_explain(sql % ('t1', 'c1,tbname')) + + def test_pipelined_agg_data_with_slimit(self): + sql_template = 'select %s from meters partition by %s %s' + + sql_elems = [ + ['count(*), min(c1), c2', 'c2', 'slimit 10', 'c2'], + ['count(*), c1, c2', 'c1, c2', 'slimit 100', 'c1,c2'], + ['count(*), c2, c1', 'c1, c2', 'slimit 1000', 'c1,c2'], + ['count(*), c4,c3', 'c3, c4', 'slimit 2000', 'c3,c4'], + ['count(*), c8,c6', 'c8, c6', 'slimit 3000', 'c8,c6'], + ['count(*), c1 +1 as a,c6', 'c1, c6', 'slimit 3000', 'a,c6'], + ['count(*), c1 +1 as a,c6', 'c1+1, c6', 'slimit 3000', 'a, c6'], + ] + + for ele in sql_elems: + sql = sql_template % (ele[0], ele[1], ele[2]) + sql_hint = self.add_hint(sql) + sql = self.add_order_by(sql, ele[3]) + sql_no_slimit = sql_template % (ele[0], ele[1], '') + sql_no_slimit = self.add_order_by(sql_no_slimit, ele[3]) + self.query_and_compare_first_rows(sql_hint, sql_no_slimit) + + def run(self): + self.prepareTestEnv() + #time.sleep(99999999) + self.test_pipelined_agg_plan_with_slimit() + self.test_pipelined_agg_data_with_slimit() + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +event = threading.Event() + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase())