Merge pull request #22730 from taosdata/fix/test_1x

test: add stream cases to 3.0
taosdata · Sep 5, 2023 · 00c5760 · 00c5760
2 parents 4a5bcb5 + db8073d
commit 00c5760
Show file tree

Hide file tree

Showing 20 changed files with 3,245 additions and 27 deletions.
diff --git a/source/dnode/vnode/src/tq/tqStreamTask.c b/source/dnode/vnode/src/tq/tqStreamTask.c
@@ -383,6 +383,9 @@ int32_t doScanWalForAllTasks(SStreamMeta* pStreamMeta, bool* pScanIdle) {
       tqDebug("s-task:%s not ready for submit block from wal, status:%s", pTask->id.idStr, pStatus);
       taosThreadMutexUnlock(&pTask->lock);
       streamMetaReleaseTask(pStreamMeta, pTask);
+      if (pItem != NULL) {
+        streamFreeQitem(pItem);
+      }
       continue;
     }
 

diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c
@@ -982,8 +982,6 @@ int32_t streamProcessDispatchRsp(SStreamTask* pTask, SStreamDispatchRsp* pRsp, i
     if (code != TSDB_CODE_SUCCESS) {  // todo: do nothing if error happens
     }
 
-    streamFreeQitem(pTask->msgInfo.pData);
-    pTask->msgInfo.pData = NULL;
     return TSDB_CODE_SUCCESS;
   }
 

diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
@@ -521,6 +521,13 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int64_t streamId, int32_t t
     ASSERT(pTask->status.timerActive == 0);
     doRemoveIdFromList(pMeta, (int32_t)taosArrayGetSize(pMeta->pTaskList), &pTask->id);
 
+    if (pTask->info.triggerParam != 0 && pTask->info.fillHistory == 0) {
+      qDebug("s-task:%s stop schedTimer, and (before) desc ref:%d", pTask->id.idStr, pTask->refCnt);
+      taosTmrStop(pTask->schedInfo.pTimer);
+      pTask->info.triggerParam = 0;
+      streamMetaReleaseTask(pMeta, pTask);
+    }
+
     streamMetaRemoveTask(pMeta, keys);
     streamMetaReleaseTask(pMeta, pTask);
   } else {

diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
@@ -6,6 +6,21 @@
 ,,y,unit-test,bash test.sh
 
 #system test
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/scalar_function.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/at_once_interval.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/at_once_session.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/at_once_state_window.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/window_close_interval.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/window_close_session.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/window_close_state_window.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/max_delay_interval.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/max_delay_session.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/at_once_interval_ext.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/max_delay_interval_ext.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/window_close_session_ext.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/partition_interval.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/pause_resume_test.py
+
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py -Q 3
@@ -24,7 +39,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_26.py -Q 2
-,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py 
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/columnLenUpdated.py -Q 4

diff --git a/tests/pytest/util/common.py b/tests/pytest/util/common.py
diff --git a/tests/pytest/util/sql.py b/tests/pytest/util/sql.py
@@ -111,7 +111,7 @@ def error(self, sql, expectedErrno = None):
             return self.error_info
 
 
-    def query(self, sql, row_tag=None,queryTimes=10):
+    def query(self, sql, row_tag=None, queryTimes=10, count_expected_res=None):
         self.sql = sql
         i=1
         while i <= queryTimes:
@@ -120,6 +120,17 @@ def query(self, sql, row_tag=None,queryTimes=10):
                 self.queryResult = self.cursor.fetchall()
                 self.queryRows = len(self.queryResult)
                 self.queryCols = len(self.cursor.description)
+
+                if count_expected_res is not None:
+                    counter = 0
+                    while count_expected_res != self.queryResult[0][0]:
+                        self.cursor.execute(sql)
+                        self.queryResult = self.cursor.fetchall()
+                        if counter < queryTimes:
+                            counter += 0.5
+                            time.sleep(0.5)
+                        else:
+                            return False
                 if row_tag:
                     return self.queryResult
                 return self.queryRows
@@ -501,15 +512,17 @@ def checkEqual(self, elm, expect_elm):
 
         caller = inspect.getframeinfo(inspect.stack()[1][0])
         args = (caller.filename, caller.lineno, self.sql, elm, expect_elm)
-        tdLog.exit("%s(%d) failed: sql:%s, elm:%s != expect_elm:%s" % args)
+        # tdLog.info("%s(%d) failed: sql:%s, elm:%s != expect_elm:%s" % args)
+        raise Exception("%s(%d) failed: sql:%s, elm:%s != expect_elm:%s" % args)
 
     def checkNotEqual(self, elm, expect_elm):
         if elm != expect_elm:
             tdLog.info("sql:%s, elm:%s != expect_elm:%s" % (self.sql, elm, expect_elm))
         else:
             caller = inspect.getframeinfo(inspect.stack()[1][0])
             args = (caller.filename, caller.lineno, self.sql, elm, expect_elm)
-            tdLog.exit("%s(%d) failed: sql:%s, elm:%s == expect_elm:%s" % args)
+            tdLog.info("%s(%d) failed: sql:%s, elm:%s == expect_elm:%s" % args)
+            raise Exception
 
     def get_times(self, time_str, precision="ms"):
         caller = inspect.getframeinfo(inspect.stack()[1][0])