From c3c8d7b07d6f57742b9123fe9d4d8288b788f5be Mon Sep 17 00:00:00 2001 From: slzhou Date: Mon, 22 Jan 2024 10:11:06 +0800 Subject: [PATCH 1/4] fix: compute scalar functions before agg in session window --- source/libs/executor/src/timewindowoperator.c | 15 +++++++++++++++ source/libs/planner/src/planSpliter.c | 5 ++--- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index b86253a8d1..59455840b0 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -30,6 +30,7 @@ typedef struct SSessionAggOperatorInfo { SOptrBasicInfo binfo; SAggSupporter aggSup; + SExprSupp scalarSupp; // supporter for perform scalar function SGroupResInfo groupResInfo; SWindowRowsSup winSup; bool reptScan; // next round scan @@ -1407,6 +1408,10 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) { } pBInfo->pRes->info.scanFlag = pBlock->info.scanFlag; + if (pInfo->scalarSupp.pExprInfo != NULL) { + SExprSupp* pExprSup = &pInfo->scalarSupp; + projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL); + } // the pDataBlock are always the same one, no need to call this again setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true); blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId); @@ -1570,6 +1575,16 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW pInfo->reptScan = false; pInfo->binfo.inputTsOrder = pSessionNode->window.node.inputTsOrder; pInfo->binfo.outputTsOrder = pSessionNode->window.node.outputTsOrder; + + if (pSessionNode->window.pExprs != NULL) { + int32_t numOfScalar = 0; + SExprInfo* pScalarExprInfo = createExprInfo(pSessionNode->window.pExprs, NULL, &numOfScalar); + code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar, &pTaskInfo->storageAPI.functionStore); + if (code != TSDB_CODE_SUCCESS) { + goto _error; + } + } + code = filterInitFromNode((SNode*)pSessionNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0); if (code != TSDB_CODE_SUCCESS) { goto _error; diff --git a/source/libs/planner/src/planSpliter.c b/source/libs/planner/src/planSpliter.c index f2258a1d7e..28e31b7a4f 100644 --- a/source/libs/planner/src/planSpliter.c +++ b/source/libs/planner/src/planSpliter.c @@ -1064,9 +1064,8 @@ static int32_t stbSplCreateMergeKeys(SNodeList* pSortKeys, SNodeList* pTargets, SNode* pTarget = NULL; bool found = false; FOREACH(pTarget, pTargets) { - if ((QUERY_NODE_COLUMN == nodeType(pSortExpr) && nodesEqualNode((SNode*)pSortExpr, pTarget)) - // || (0 == strcmp(pSortExpr->aliasName, ((SColumnNode*)pTarget)->colName)) - ) { + if ((QUERY_NODE_COLUMN == nodeType(pSortExpr) && nodesEqualNode((SNode*)pSortExpr, pTarget)) || + (0 == strcmp(pSortExpr->aliasName, ((SColumnNode*)pTarget)->colName))) { code = nodesListMakeStrictAppend(&pMergeKeys, stbSplCreateOrderByExpr(pSortKey, pTarget)); if (TSDB_CODE_SUCCESS != code) { break; From 0e525a8e224e86e6d4bf9fa92c6b651aefefc445 Mon Sep 17 00:00:00 2001 From: slzhou Date: Thu, 18 Jan 2024 14:57:12 +0800 Subject: [PATCH 2/4] fix: add test case --- tests/parallel_test/cases.task | 1 + tests/system-test/2-query/partition_expr.py | 35 +++++++++++++++++++++ 2 files changed, 36 insertions(+) create mode 100644 tests/system-test/2-query/partition_expr.py diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 0687a3271c..8eeecce750 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -43,6 +43,7 @@ fi #,,n,system-test,python3 ./test.py -f 8-stream/snode_restart.py -N 4 ,,n,system-test,python3 ./test.py -f 8-stream/snode_restart_with_checkpoint.py -N 4 +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/partition_expr.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/project_group.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tbname_vgroup.py ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/count_interval.py diff --git a/tests/system-test/2-query/partition_expr.py b/tests/system-test/2-query/partition_expr.py new file mode 100644 index 0000000000..c03d7eccb3 --- /dev/null +++ b/tests/system-test/2-query/partition_expr.py @@ -0,0 +1,35 @@ +from wsgiref.headers import tspecials +from util.log import * +from util.cases import * +from util.sql import * +import numpy as np + + +class TDTestCase: + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor()) + + self.rowNum = 10 + self.batchNum = 5 + self.ts = 1537146000000 + + def run(self): + dbname = "db" + tdSql.prepare() + + tdSql.execute(f'''create table sta(ts timestamp, f int, col2 bigint) tags(tg1 int, tg2 binary(20))''') + tdSql.execute(f"create table sta1 using sta tags(1, 'a')") + tdSql.execute(f"insert into sta1 values(1537146000001, 11, 110)") + tdSql.execute(f"insert into sta1 values(1537146000002, 12, 120)") + tdSql.execute(f"insert into sta1 values(1537146000003, 13, 130)") + + tdSql.query("select _wstart, f+100, count(*) from db.sta partition by f+100 session(ts, 1a) order by _wstart"); + tdSql.checkData(0, 1, 111.0) + def stop(self): + tdSql.close() + tdLog.success("%s successfully executed" % __file__) + +tdCases.addWindows(__file__, TDTestCase()) +tdCases.addLinux(__file__, TDTestCase()) From 5021d981b8f4c3f3b1b3630b03644e8532355d37 Mon Sep 17 00:00:00 2001 From: slzhou Date: Mon, 22 Jan 2024 11:23:38 +0800 Subject: [PATCH 3/4] fix: fix memory leak --- source/libs/executor/src/timewindowoperator.c | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index 59455840b0..d90fdb38df 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1536,6 +1536,8 @@ void destroySWindowOperatorInfo(void* param) { colDataDestroy(&pInfo->twAggSup.timeWindowData); cleanupAggSup(&pInfo->aggSup); + cleanupExprSupp(&pInfo->scalarSupp); + cleanupGroupResInfo(&pInfo->groupResInfo); taosMemoryFreeClear(param); } @@ -1592,7 +1594,8 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW setOperatorInfo(pOperator, "SessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, true, OP_NOT_OPENED, pInfo, pTaskInfo); - pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doSessionWindowAgg, NULL, destroySWindowOperatorInfo, + pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, cleanupExprSupp(&pInfo->scalarSupp); +WindowAgg, NULL, destroySWindowOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); pOperator->pTaskInfo = pTaskInfo; code = appendDownstream(pOperator, &downstream, 1); From 60b3ac93abe32363b8786bb373a17cc5a0b6b5e8 Mon Sep 17 00:00:00 2001 From: slzhou Date: Mon, 22 Jan 2024 13:11:11 +0800 Subject: [PATCH 4/4] fix: fix compilation error --- source/libs/executor/src/timewindowoperator.c | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index d90fdb38df..57c038e75a 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -1594,8 +1594,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionW setOperatorInfo(pOperator, "SessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, true, OP_NOT_OPENED, pInfo, pTaskInfo); - pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, cleanupExprSupp(&pInfo->scalarSupp); -WindowAgg, NULL, destroySWindowOperatorInfo, + pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doSessionWindowAgg, NULL, destroySWindowOperatorInfo, optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL); pOperator->pTaskInfo = pTaskInfo; code = appendDownstream(pOperator, &downstream, 1);