feat: [TS-6100] Forbid where when using %%trows (#31827)

* feat: [TS-6100] Forbid where when using %%trows

* test: update cases

* feat: [TS-6100] Fix leaks.

---------

Co-authored-by: Simon Guan <[email protected]>

Author: Simon9997

source/libs/parser/src/parTranslater.c

@@ -3053,16 +3053,7 @@ static EDealRes translatePlaceHolderFunc(STranslateContext* pCxt, SNode** pFunc)
     }
     case FUNCTION_TYPE_PLACEHOLDER_TBNAME: {
       BIT_FLAG_SET_MASK(pCxt->placeHolderBitmap, PLACE_HOLDER_PARTITION_TBNAME);
-      if (BIT_FLAG_TEST_MASK(pCxt->placeHolderBitmap, PLACE_HOLDER_PARTITION_ROWS) && pCxt->createStreamCalc) {
-        SFunctionNode *pTbname = NULL;
-        PAR_ERR_JRET(createTbnameFunction(&pTbname));
-        tstrncpy(pTbname->node.userAlias, ((SExprNode*)*pFunc)->userAlias, TSDB_COL_NAME_LEN);
-        nodesDestroyNode(*pFunc);
-        *pFunc = (SNode*)pTbname;
-        return translateFunction(pCxt, (SFunctionNode**)pFunc);
-      } else {
-        PAR_ERR_JRET(nodesMakeValueNodeFromString("", (SValueNode**)&extraValue));
-      }
+      PAR_ERR_JRET(nodesMakeValueNodeFromString("", (SValueNode**)&extraValue));
       break;
     }
     case FUNCTION_TYPE_PLACEHOLDER_COLUMN: {
@@ -3083,32 +3074,11 @@ static EDealRes translatePlaceHolderFunc(STranslateContext* pCxt, SNode** pFunc)
         PAR_ERR_JRET(generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_STREAM_INVALID_PLACE_HOLDER, "%%n : partition index out of range"));
       }
 
-      if (BIT_FLAG_TEST_MASK(pCxt->placeHolderBitmap, PLACE_HOLDER_PARTITION_ROWS) && pCxt->createStreamCalc) {
-        if (nodeType(pExpr) == QUERY_NODE_FUNCTION) {
-          SFunctionNode* pTbname = NULL;
-          PAR_ERR_JRET(createTbnameFunction(&pTbname));
-          tstrncpy(pTbname->node.userAlias, ((SExprNode*)*pFunc)->userAlias, TSDB_COL_NAME_LEN);
-          nodesDestroyNode(*pFunc);
-          *pFunc = (SNode*)pTbname;
-          return translateFunction(pCxt, (SFunctionNode**)pFunc);
-        } else if (nodeType(pExpr) == QUERY_NODE_COLUMN) {
-          SColumnNode* pCol = NULL;
-          PAR_ERR_JRET(nodesCloneNode((SNode*)pExpr, (SNode**)&pCol));
-          tstrncpy(pCol->node.userAlias, ((SExprNode*)*pFunc)->userAlias, TSDB_COL_NAME_LEN);
-          nodesDestroyNode(*pFunc);
-          *pFunc = (SNode*)pCol;
-          return translateColumn(pCxt, (SColumnNode**)pFunc);
-        } else {
-          PAR_ERR_JRET(generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_STREAM_INVALID_PLACE_HOLDER,
-                                           "%%n : partition index must be a column or tbname function"));
-        }
-      } else {
-        PAR_ERR_JRET(nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&extraValue));
-        ((SValueNode*)extraValue)->node.resType = pExpr->resType;
-        ((SValueNode*)extraValue)->isNull = true;
+      PAR_ERR_JRET(nodesMakeNode(QUERY_NODE_VALUE, (SNode**)&extraValue));
+      ((SValueNode*)extraValue)->node.resType = pExpr->resType;
+      ((SValueNode*)extraValue)->isNull = true;
 
-        pFuncNode->node.resType = pExpr->resType;
-      }
+      pFuncNode->node.resType = pExpr->resType;
       break;
     }
     default:
@@ -4746,7 +4716,7 @@ static int32_t setTrowsTableVgroupList(STranslateContext* pCxt, SName* pName, SR
   if (vg == NULL) {
     PAR_ERR_JRET(terrno);
   }
-  pRealTable->pVgroupList->vgroups[0] = *vg;
+  memcpy(pRealTable->pVgroupList->vgroups, vg, sizeof(SVgroupInfo));
 
 _return:
   taosArrayDestroy(vgroupList);
@@ -6040,10 +6010,14 @@ static int32_t translatePlaceHolderTable(STranslateContext* pCxt, SNode** pTable
           newPlaceHolderTable->pMeta->tableType == TSDB_SUPER_TABLE) {
         newPlaceHolderTable->asSingleTable = true;
       }
+      if (inJoin) {
+        PAR_ERR_JRET(generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
+                                             "%%%%trows should not appear in join condition"));
+      }
       break;
     }
     default: {
-      PAR_ERR_JRET(generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_TSC_INVALID_OPERATION, "invalid placeholder table type"));
+      PAR_ERR_JRET(generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "invalid placeholder table type"));
       break;
     }
   }
@@ -6052,6 +6026,7 @@ static int32_t translatePlaceHolderTable(STranslateContext* pCxt, SNode** pTable
 
   return code;
 _return:
+  nodesDestroyNode((SNode*)newPlaceHolderTable);
   parserError("translatePlaceHolderTable failed, code:%d, errmsg:%s", code, tstrerror(code));
   return code;
 }
@@ -8395,10 +8370,13 @@ static int32_t setTableVgroupsFromEqualTbnameCond(STranslateContext* pCxt, SSele
 
 static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) {
   pCxt->currClause = SQL_CLAUSE_WHERE;
-  int32_t code = translateExpr(pCxt, &pSelect->pWhere);
-  if (TSDB_CODE_SUCCESS == code) {
-    code = getQueryTimeRange(pCxt, &pSelect->pWhere, &pSelect->timeRange, &pSelect->pTimeRange, pSelect->pFromTable);
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (pSelect->pWhere && BIT_FLAG_TEST_MASK(pCxt->placeHolderBitmap, PLACE_HOLDER_PARTITION_ROWS) && pCxt->createStreamCalc) {
+    PAR_ERR_RET(generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY,
+                                       "%%%%trows can not be used with WHERE clause."));
   }
+  PAR_ERR_RET(translateExpr(pCxt, &pSelect->pWhere));
+  PAR_ERR_RET(getQueryTimeRange(pCxt, &pSelect->pWhere, &pSelect->timeRange, &pSelect->pTimeRange, pSelect->pFromTable));
   if (pSelect->pWhere != NULL && pCxt->pParseCxt->topicQuery == false) {
     PAR_ERR_RET(setTableVgroupsFromEqualTbnameCond(pCxt, pSelect));
   }

test/cases/13-StreamProcessing/07-SubQuery/test_subquery_count_1.py

@@ -206,7 +206,7 @@ def createStreams(self):
 
         stream = StreamItem(
             id=6,
-            stream="create stream rdb.s6 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r6 as select _twstart ts, count(c1), avg(c2) from %%trows where ts >= _twstart and ts <= _twend partition by tbname",
+            stream="create stream rdb.s6 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r6 as select _twstart ts, count(c1), avg(c2) from %%trows partition by tbname",
             res_query="select *, tag_tbname from rdb.r6 where tag_tbname='t1'",
             exp_query="select _wstart, count(c1), avg(c2), 't1', 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00' and ts < '2025-01-01 00:35:00' interval(5m);",
         )
@@ -222,7 +222,7 @@ def createStreams(self):
 
         stream = StreamItem(
             id=8,
-            stream="create stream rdb.s8 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r8 as select _twstart ts, count(c1), avg(c2) from %%trows where ts >= _twstart and ts <= _twend partition by %%1",
+            stream="create stream rdb.s8 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r8 as select _twstart ts, count(c1), avg(c2) from %%trows partition by %%1",
             res_query="select *, tag_tbname from rdb.r8 where tag_tbname='t1'",
             exp_query="select _wstart, count(c1), avg(c2), 't1', 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00' and ts < '2025-01-01 00:35:00' interval(5m);",
         )
@@ -256,9 +256,9 @@ def createStreams(self):
 
         stream = StreamItem(
             id=12,
-            stream="create stream rdb.s12 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r12 as select _twstart ts, %%tbname tb, %%1, count(*) v1, avg(c1) v2, first(c1) v3, last(c1) v4 from %%trows where c2 > 0;",
+            stream="create stream rdb.s12 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r12 as select _twstart ts, %%tbname tb, %%1, count(*) v1, avg(c1) v2, first(c1) v3, last(c1) v4 from %%trows;",
             res_query="select ts, tb, `%%1`, v2, v3, v4, tag_tbname from rdb.r12 where tb='t1'",
-            exp_query="select _wstart, 't1', 't1', avg(c1) v2, first(c1) v3, last(c1) v4, 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00' and ts < '2025-01-01 00:35:00' and c2 > 0 interval(5m);",
+            exp_query="select _wstart, 't1', 't1', avg(c1) v2, first(c1) v3, last(c1) v4, 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00' and ts < '2025-01-01 00:35:00' interval(5m);",
             check_func=self.check12,
         )
         self.streams.append(stream)
@@ -435,9 +435,9 @@ def createStreams(self):
 
         stream = StreamItem(
             id=34,
-            stream="create stream rdb.s34 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r34 as select _twstart - 5m tc, TIMEDIFF(_twstart - 5m, _twstart + 5m) tx, %%tbname tb, %%1 tg1, sum(c1) c1, avg(c2) c2, first(c1) c3, last(c2) c4 from %%trows where c1 > 0;",
+            stream="create stream rdb.s34 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r34 as select _twstart - 5m tc, TIMEDIFF(_twstart - 5m, _twstart + 5m) tx, %%tbname tb, %%1 tg1, sum(c1) c1, avg(c2) c2, first(c1) c3, last(c2) c4 from %%trows;",
             res_query="select * from rdb.r34 where tag_tbname = 't1';",
-            exp_query="select _wstart - 5m, TIMEDIFF(_wstart, _wend) * 2, 't1', 't1', sum(c1) c1, avg(c2) c2, first(c1) c3, last(c2) c4 , 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:35:00.000' and c1 > 0 interval(5m);",
+            exp_query="select _wstart - 5m, TIMEDIFF(_wstart, _wend) * 2, 't1', 't1', sum(c1) c1, avg(c2) c2, first(c1) c3, last(c2) c4 , 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:35:00.000' interval(5m);",
         )
         self.streams.append(stream)
 
@@ -613,9 +613,9 @@ def createStreams(self):
 
         stream = StreamItem(
             id=56,
-            stream="create stream rdb.s56 count_window(1, c1) from tdb.v1 into rdb.r56 as select _wstart ws, _wend we, _twstart tws, _twend + 5m twe, first(c1) cf, last(c1) cl, count(c1) cc from %%trows where ts >= _twstart and ts < _twend + 5m interval(1m) fill(prev)",
+            stream="create stream rdb.s56 count_window(1, c1) from tdb.v1 into rdb.r56 as select _wstart ws, _wend we, _twstart tws, _twend + 5m twe, first(c1) cf, last(c1) cl, count(c1) cc from %%trows interval(1m)",
             res_query="select * from rdb.r56 where ws >= '2025-01-01 00:00:00.000' and we <= '2025-01-01 00:05:00.000' ",
-            exp_query="select _wstart ws, _wend we, cast('2025-01-01 00:00:00.000' as timestamp) tws, cast('2025-01-01 00:05:00.000' as timestamp) twe, first(c1) cf, last(c1) cl, count(c1) cc from tdb.v1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:05:00.000' interval(1m) fill(prev);",
+            exp_query="select _wstart ws, _wend we, cast('2025-01-01 00:00:00.000' as timestamp) tws, cast('2025-01-01 00:05:00.000' as timestamp) twe, first(c1) cf, last(c1) cl, count(c1) cc from tdb.v1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:05:00.000' interval(1m);",
         )
         self.streams.append(stream)
 
@@ -878,7 +878,7 @@ def createStreams(self):
 
         stream = StreamItem(
             id=89,
-            stream="create stream rdb.s89 interval(5m) sliding(5m) from tdb.v1 into rdb.r89 as select _twstart tw, _c0 ta, _rowts tb, c1, c2, rand() c3 from %%trows where _c0 >= _twstart and _c0 < _twend + 5m order by _c0 limit 1",
+            stream="create stream rdb.s89 interval(5m) sliding(5m) from tdb.v1 into rdb.r89 as select _twstart tw, _c0 ta, _rowts tb, c1, c2, rand() c3 from %%trows order by _c0 limit 1",
             res_query="select tw, ta, tb, c1, c2 from rdb.r89 limit 3",
             exp_query="select _c0,  _c0, _rowts, c1, c2 from tdb.v1 where ts = '2025-01-01 00:00:00.000' or ts = '2025-01-01 00:05:00.000' or ts = '2025-01-01 00:10:00.000' order by _c0 limit 3;",
             check_func=self.check89,
@@ -895,7 +895,7 @@ def createStreams(self):
 
         stream = StreamItem(
             id=91,
-            stream="create stream rdb.s91 count_window(1, c1) from tdb.triggers partition by id, name, tbname into rdb.r91 as select _twstart, id cid, name cname, sum(c1) amount from %%trows where ts between _twstart and _twend + 5m and id=%%1 and name=%%1 partition by id, name having sum(c1) <= 5;",
+            stream="create stream rdb.s91 count_window(1, c1) from tdb.triggers partition by id, name, tbname into rdb.r91 as select _twstart, id cid, name cname, sum(c1) amount from %%trows partition by id, name having sum(c1) <= 5;",
             res_query="select * from rdb.r91",
             exp_query="select _wstart, id, name, sum(c1), id, name, tbname from tdb.t1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:15:00.000' partition by tbname interval(5m) having sum(c1) <= 5;",
         )
@@ -1159,9 +1159,9 @@ def createStreams(self):
 
         stream = StreamItem(
             id=124,
-            stream="create stream rdb.s124 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r124 as select _wstart ts, count(c1), sum(c2) from %%trows where ts >= _twstart and ts <= _twend + 5m interval(1m) fill(prev)",
+            stream="create stream rdb.s124 count_window(1, c1) from tdb.triggers partition by tbname into rdb.r124 as select _wstart ts, count(c1), sum(c2) from %%trows interval(1m)",
             res_query="select * from rdb.r124 where tag_tbname='t1' limit 15;",
-            exp_query="select _wstart, count(c1), sum(c2), 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:15:00.000' interval(1m) fill(prev);",
+            exp_query="select _wstart, count(c1), sum(c2), 't1' from tdb.t1 where ts >= '2025-01-01 00:00:00.000' and ts < '2025-01-01 00:35:00.000' interval(1m);",
         )
         self.streams.append(stream)
 
@@ -1219,7 +1219,7 @@ def createStreams(self):
             res_query="select tats, tbts, tac1, tac2, tbc1, tbc2 from rdb.r131",
             exp_query="select ta.ts tats, tb.cts tbts, ta.c1 tac1, ta.c2 tac2, tb.cint tbc1, tb.cuint tbc2 from tdb.v1 ta right join qdb.t1 tb on ta.ts=tb.cts where ta.ts >= '2025-01-01 00:00:00.000' and ta.ts < '2025-01-01 00:35:00.000';",
         )
-        # self.streams.append(stream) TD-36477
+        # self.streams.append(stream) forbidden
 
         stream = StreamItem(
             id=132,
@@ -1234,7 +1234,7 @@ def createStreams(self):
             stream="create stream rdb.s133 count_window(1) from tdb.v2 partition by id, tbname into rdb.r133 as select ta.ts tats, tb.cts tbts, ta.c1 tac1, ta.c2 tac2, tb.cint tbc1, tb.cuint tbc2, _twstart, _twend from %%trows ta right join qdb.t1 tb on ta.ts=tb.cts where ta.ts >= _twstart and ta.ts < _twend + 5m;",
             check_func=self.check133,
         )
-        self.streams.append(stream)
+        # self.streams.append(stream) forbidden
 
         stream = StreamItem(
             id=134,