fix: windows pList not locked and add more info to minidump file (#35080)

DuanKuanJun · web-flow · commit b64bdd2f26e3 · 2026-04-15T09:10:39.000+08:00
diff --git a/cmake/define.cmake b/cmake/define.cmake
@@ -145,7 +145,16 @@ IF(TD_WINDOWS)
         # /Zi  : generate a separate PDB file (previously /Zi- which disabled it entirely).
         # The PDB is NOT shipped to the user but must be archived internally per version
         # so that crash dumps from the field can be symbolicated with WinDbg / VS.
-        SET(COMMON_FLAGS "/W3 /D_WIN32 /DWIN32 /Zi /O2 /GL /MD")
+        IF(${BUILD_SANITIZER})
+            MESSAGE("${Green} will build with AddressSanitizer (MSVC ASan)! ${ColourReset}")
+            # /fsanitize=address: MSVC ASan (incompatible with /GL and /O2)
+            # _DISABLE_VECTOR_ANNOTATION/_DISABLE_STRING_ANNOTATION: suppress STL ASan
+            # annotations to avoid LNK2038 mismatch with pre-built libs (e.g. rocksdb)
+            # that were compiled without /fsanitize=address.
+            SET(COMMON_FLAGS "/W3 /D_WIN32 /DWIN32 /Zi /O1 /MD /fsanitize=address /D_DISABLE_VECTOR_ANNOTATION=1 /D_DISABLE_STRING_ANNOTATION=1")
+        ELSE()
+            SET(COMMON_FLAGS "/W3 /D_WIN32 /DWIN32 /Zi /O2 /GL /MD")
+        ENDIF()
     ELSE()
         MESSAGE("${Green} will build Debug version! ${ColourReset}")
         # NOTE: let cmake to choose default compile options
diff --git a/source/dnode/mgmt/exe/dmMain.c b/source/dnode/mgmt/exe/dmMain.c
@@ -190,6 +190,7 @@ void dmLogCrash(int signum, void *sigInfo, void *context) {
   // taosIgnSignal(SIGHUP);
   // taosIgnSignal(SIGINT);
   // taosIgnSignal(SIGBREAK);
+  dInfo("crash signal is %d", signum);
 
 #ifndef WINDOWS
   if (taosIgnSignal(SIGBUS) != 0) {
@@ -200,17 +201,22 @@ void dmLogCrash(int signum, void *sigInfo, void *context) {
     dWarn("failed to ignore signal SIGABRT");
   }
   if (taosIgnSignal(SIGFPE) != 0) {
-    dWarn("failed to ignore signal SIGABRT");
+    dWarn("failed to ignore signal SIGFPE");
   }
   if (taosIgnSignal(SIGSEGV) != 0) {
-    dWarn("failed to ignore signal SIGABRT");
+    dWarn("failed to ignore signal SIGSEGV");
   }
 #ifdef USE_REPORT
   writeCrashLogToFile(signum, sigInfo, CUS_PROMPT "d", dmGetClusterId(), global.startTime);
 #endif
 #ifdef _TD_DARWIN_64
   exit(signum);
 #elif defined(WINDOWS)
+  // On Windows, restore default signal handler and re-raise to trigger SEH/FlCrashDump
+  // This allows the UnhandledExceptionFilter to generate a proper minidump
+  signal(signum, SIG_DFL);
+  raise(signum);
+  // If raise() returns (shouldn't happen), fall through to exit
   exit(signum);
 #endif
 }
diff --git a/source/dnode/mnode/impl/src/mndStreamMgmt.c b/source/dnode/mnode/impl/src/mndStreamMgmt.c
@@ -768,7 +768,13 @@ int32_t msmBuildReaderDeployInfo(SStmTaskDeploy* pDeploy, void* calcScanPlan, SS
   } else {
     SStreamReaderDeployFromCalc* pCalc = &pMsg->msg.calc;
     pCalc->execReplica = pInfo->runnerDeploys * pInfo->runnerReplica;
-    pCalc->calcScanPlan = calcScanPlan;
+    if (calcScanPlan) {
+      pCalc->calcScanPlan = taosStrdup(calcScanPlan);
+      if (NULL == pCalc->calcScanPlan) {
+        return terrno;
+      }
+      pCalc->freeScanPlan = true;
+    }
   }
 
   return TSDB_CODE_SUCCESS;
@@ -2026,14 +2032,19 @@ static int32_t msmUpdateCalcReaderTasks(SStreamObj* pStream, SNodeList* pSubEP)
 
     for (int32_t i = 0; i < taskNum; ++i) {
       SStmTaskToDeployExt* pExt = taosArrayGet(pVg->taskList, i);
-      if (pExt->deploy.task.streamId != streamId || STREAM_READER_TASK != pExt->deploy.task.type) {
+      if (pExt->deployed || pExt->deploy.task.streamId != streamId || STREAM_READER_TASK != pExt->deploy.task.type) {
         continue;
       }
 
       if (!pExt->deploy.msg.reader.triggerReader) {
         SStreamReaderDeployFromCalc* pCalcReaderDeploy = &pExt->deploy.msg.reader.msg.calc;
         TAOS_CHECK_EXIT(nodesStringToNode(pCalcReaderDeploy->calcScanPlan, (SNode**)&pSubplan));
         TAOS_CHECK_EXIT(nodesCloneList(pSubEP, &pSubplan->pSubQ));
+        
+        // Free old calcScanPlan before nodesNodeToString overwrites the pointer
+        if (pCalcReaderDeploy->freeScanPlan) {
+          taosMemoryFreeClear(pCalcReaderDeploy->calcScanPlan);
+        }
         TAOS_CHECK_EXIT(nodesNodeToString((SNode*)pSubplan, false, (char**)&pCalcReaderDeploy->calcScanPlan, NULL));
         pCalcReaderDeploy->freeScanPlan = true;
         nodesDestroyNode((SNode *)pSubplan);
diff --git a/source/dnode/mnode/impl/src/mndStreamUtil.c b/source/dnode/mnode/impl/src/mndStreamUtil.c
@@ -75,11 +75,13 @@ void mstDestroySStmTaskToDeployExt(void* param) {
     case STREAM_READER_TASK:
       if (!pExt->deploy.msg.reader.triggerReader) {
         SStreamReaderDeployFromCalc* pCalcReaderDeploy = &pExt->deploy.msg.reader.msg.calc;
-        taosMemoryFreeClear(pCalcReaderDeploy->calcScanPlan);
+        if (pCalcReaderDeploy->freeScanPlan) {
+          taosMemoryFreeClear(pCalcReaderDeploy->calcScanPlan);
+        }
       }
       break;
     default:  
-      break;;
+      break;
   }
 }
 
diff --git a/source/libs/decimal/src/detail/CMakeLists.txt b/source/libs/decimal/src/detail/CMakeLists.txt
@@ -4,6 +4,16 @@ aux_source_directory(. WIDE_INTEGER_SRC)
 SET(CMAKE_CXX_STANDARD 14)
 add_library(wideInteger STATIC ${WIDE_INTEGER_SRC})
 
+# When building with MSVC ASan (/fsanitize=address), suppress STL container
+# annotations to avoid LNK2038 mismatch with pre-built libs (e.g. rocksdb)
+# that were compiled without /fsanitize=address.
+if(TD_WINDOWS)
+    target_compile_definitions(wideInteger PRIVATE
+        _DISABLE_VECTOR_ANNOTATION=1
+        _DISABLE_STRING_ANNOTATION=1
+    )
+endif()
+
 target_include_directories(
     wideInteger
     PUBLIC "${TD_SOURCE_DIR}/source/libs/decimal/inc/"
diff --git a/source/libs/new-stream/inc/streamTriggerTask.h b/source/libs/new-stream/inc/streamTriggerTask.h
@@ -224,11 +224,12 @@ typedef struct SSTriggerRealtimeContext {
   SSHashObj              *pGroupColVals;  // SSHashObj<gid, SArray<SStreamGroupValue>*>
 
   // these fields need to be cleared each round
+  bool       needCheckAgain;
   SSHashObj *pSlices;  // SSHashObj<uid, SSTriggerDataSlice>
+  SObjList   dumpTableUids;  // SObjList<{uid, vgId}>, backup ids for repeated check in one round
   // these fields are shared by all groups and need to reset for each group
   bool                         needPseudoCols;
   bool                         needMergeWindow;
-  bool                         needCheckAgain;
   SSTriggerNewTimestampSorter *pSorter;
   SSTriggerNewVtableMerger    *pMerger;
   SArray                      *pParentWindows;  // SArray<SSTriggerNotifyWindow>, valid parent windows in this round
diff --git a/source/libs/new-stream/src/streamTriggerTask.c b/source/libs/new-stream/src/streamTriggerTask.c
@@ -3618,6 +3618,8 @@ static int32_t stRealtimeContextInit(SSTriggerRealtimeContext *pContext, SStream
 
   pContext->pSlices = tSimpleHashInit(256, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT));
   QUERY_CHECK_NULL(pContext->pSlices, code, lino, _end, terrno);
+  code = taosObjListInit(&pContext->dumpTableUids, &pContext->tableUidPool);
+  QUERY_CHECK_CODE(code, lino, _end);
   if (pTask->isVirtualTable) {
     code = stRealtimeContextInitPatchContext(pContext);
     QUERY_CHECK_CODE(code, lino, _end);
@@ -3810,6 +3812,7 @@ static void stRealtimeContextDestroy(void *ptr) {
     tSimpleHashCleanup(pContext->pSlices);
     pContext->pSlices = NULL;
   }
+  taosObjListClear(&pContext->dumpTableUids);
   stRealtimeContextDestroyPatchContext(pContext);
 
   if (pContext->pSorter != NULL) {
@@ -4340,7 +4343,7 @@ static int32_t stRealtimeContextSendPullReq(SSTriggerRealtimeContext *pContext,
 
 // send STRIGGER_PULL_GROUP_COL_VALUE for given (pProgress, gid); used when pulling groupInfo for pending create-table
 static int32_t stRealtimeContextSendPullReqForGid(SSTriggerRealtimeContext *pContext, SSTriggerWalProgress *pProgress,
-                                                   int64_t gid) {
+                                                  int64_t gid) {
   SStreamTriggerTask       *pTask = pContext->pTask;
   SSTriggerPullRequest      *pReq = &pProgress->pullReq.base;
   SStreamTaskAddr           *pReader = pProgress->pTaskAddr;
@@ -5211,6 +5214,31 @@ static int32_t stRealtimeContextCheckIdleGroup(SSTriggerRealtimeContext *pContex
   return code;
 }
 
+static int32_t stRealtimeContextCopyTableUids(SSTriggerRealtimeContext *pContext, SObjList *pSrc, SObjList *pDst) {
+  int32_t             code = TSDB_CODE_SUCCESS;
+  int32_t             lino = 0;
+  int64_t            *id = NULL;
+  SObjListIter        iter = {0};
+  SStreamTriggerTask *pTask = pContext->pTask;
+
+  if (pSrc == NULL || pDst == NULL || pSrc->neles == 0) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  taosObjListClear(pDst);
+  taosObjListInitIter(pSrc, &iter, TOBJLIST_ITER_FORWARD);
+  while ((id = taosObjListIterNext(&iter)) != NULL) {
+    code = taosObjListAppend(pDst, id);
+    QUERY_CHECK_CODE(code, lino, _end);
+  }
+
+_end:
+  if (code != TSDB_CODE_SUCCESS) {
+    ST_TASK_ELOG("%s failed at line %d since %s", __func__, lino, tstrerror(code));
+  }
+  return code;
+}
+
 static int32_t stRealtimeContextCheck(SSTriggerRealtimeContext *pContext) {
   int32_t             code = TSDB_CODE_SUCCESS;
   int32_t             lino = 0;
@@ -5371,20 +5399,20 @@ static int32_t stRealtimeContextCheck(SSTriggerRealtimeContext *pContext) {
     }
   }
 
-  // Check for idle groups
-  code = stRealtimeContextCheckIdleGroup(pContext);
-  QUERY_CHECK_CODE(code, lino, _end);
-
-  while (TD_DLIST_NELES(&pContext->groupsToCheck) > 0) {
+_check:
+  while (TD_DLIST_NELES(&pContext->groupsToCheck) > 0 && pContext->status != STRIGGER_CONTEXT_ACQUIRE_REQUEST &&
+         pContext->status != STRIGGER_CONTEXT_SEND_CALC_REQ) {
     SSTriggerRealtimeGroup *pGroup = TD_DLIST_HEAD(&pContext->groupsToCheck);
     switch (pContext->status) {
       case STRIGGER_CONTEXT_FETCH_META: {
-        pContext->status = STRIGGER_CONTEXT_ACQUIRE_REQUEST;
-      }
-      case STRIGGER_CONTEXT_ACQUIRE_REQUEST: {
         pContext->status = STRIGGER_CONTEXT_CHECK_CONDITION;
       }
       case STRIGGER_CONTEXT_CHECK_CONDITION: {
+        if (pTask->triggerType == STREAM_TRIGGER_SLIDING) {
+          // only sliding trigger may need to check again
+          code = stRealtimeContextCopyTableUids(pContext, &pGroup->tableUids, &pContext->dumpTableUids);
+          QUERY_CHECK_CODE(code, lino, _end);
+        }
         code = stRealtimeGroupCheck(pGroup);
         QUERY_CHECK_CODE(code, lino, _end);
         if (pContext->needPseudoCols) {
@@ -5409,18 +5437,29 @@ static int32_t stRealtimeContextCheck(SSTriggerRealtimeContext *pContext) {
         QUERY_CHECK_CODE(code, lino, _end);
       }
     }
-    if (!pContext->needCheckAgain) {
-      stRealtimeGroupClearMetadatas(pGroup);
-      TD_DLIST_POP(&pContext->groupsToCheck, pGroup);
-      code = stTriggerTaskReadyRecalcRequest(pTask, pGroup);
+    if (pContext->needCheckAgain && pTask->triggerType == STREAM_TRIGGER_SLIDING) {
+      code = stRealtimeContextCopyTableUids(pContext, &pContext->dumpTableUids, &pGroup->tableUids);
       QUERY_CHECK_CODE(code, lino, _end);
+    } else {
+      TD_DLIST_POP(&pContext->groupsToCheck, pGroup);
+      stRealtimeGroupClearMetadatas(pGroup);
+    }
+    pContext->needCheckAgain = false;
+    taosObjListClear(&pContext->dumpTableUids);
+    code = stTriggerTaskReadyRecalcRequest(pTask, pGroup);
+    QUERY_CHECK_CODE(code, lino, _end);
+    pContext->status = STRIGGER_CONTEXT_FETCH_META;
+    if (pGroup->pPendingCalcParams.neles >= STREAM_CALC_REQ_MAX_WIN_NUM ||
+        pContext->calcParamPool.size >= STREAM_TRIGGER_MAX_PENDING_PARAMS) {
+      break;
     }
-    pContext->status = STRIGGER_CONTEXT_ACQUIRE_REQUEST;
   }
 
   if (pContext->pMinGroup == NULL && pContext->pMaxDelayHeap->min != NULL) {
     pContext->pMinGroup = container_of(pContext->pMaxDelayHeap->min, SSTriggerRealtimeGroup, heapNode);
-    if (pContext->pMinGroup->nextExecTime > now) {
+    if (pContext->pMinGroup->nextExecTime > now &&
+        pContext->pMinGroup->pPendingCalcParams.neles < STREAM_CALC_REQ_MAX_WIN_NUM &&
+        pContext->calcParamPool.size < STREAM_TRIGGER_MAX_PENDING_PARAMS) {
       pContext->pMinGroup = NULL;
     }
   }
@@ -5465,7 +5504,7 @@ static int32_t stRealtimeContextCheck(SSTriggerRealtimeContext *pContext) {
             // calc req has not been set
             goto _end;
           }
-          if (pTask->placeHolderBitmap & PLACE_HOLDER_PARTITION_ROWS) {
+          if (pTask->placeHolderBitmap & PLACE_HOLDER_PARTITION_ROWS && !IS_TRIGGER_GROUP_TO_CHECK(pGroup)) {
             stRealtimeGroupClearMetadatas(pGroup);
           }
           stRealtimeGroupClearTempState(pGroup);
@@ -5490,7 +5529,9 @@ static int32_t stRealtimeContextCheck(SSTriggerRealtimeContext *pContext) {
     pContext->status = STRIGGER_CONTEXT_ACQUIRE_REQUEST;
     if (pContext->pMaxDelayHeap->min != NULL) {
       pContext->pMinGroup = container_of(pContext->pMaxDelayHeap->min, SSTriggerRealtimeGroup, heapNode);
-      if (pContext->pMinGroup->nextExecTime > now) {
+      if (pContext->pMinGroup->nextExecTime > now &&
+          pContext->pMinGroup->pPendingCalcParams.neles < STREAM_CALC_REQ_MAX_WIN_NUM &&
+          pContext->calcParamPool.size < STREAM_TRIGGER_MAX_PENDING_PARAMS) {
         pContext->pMinGroup = NULL;
       }
     } else {
@@ -5507,6 +5548,11 @@ static int32_t stRealtimeContextCheck(SSTriggerRealtimeContext *pContext) {
     }
   }
 
+  if (TD_DLIST_NELES(&pContext->groupsToCheck) > 0) {
+    pContext->status = STRIGGER_CONTEXT_CHECK_CONDITION;
+    goto _check;
+  }
+
   int32_t deleteGroupNum = taosArrayGetSize(pContext->groupsToDelete);
   if (deleteGroupNum > 0) {
     pContext->status = STRIGGER_CONTEXT_SEND_DROP_REQ;
@@ -6121,6 +6167,9 @@ static int32_t stRealtimeContextProcPullRsp(SSTriggerRealtimeContext *pContext,
       if (latestVersionTime != INT64_MAX) {
         atomic_store_64(&pTask->latestVersionTime, latestVersionTime);
       }
+      // Check for idle groups
+      code = stRealtimeContextCheckIdleGroup(pContext);
+      QUERY_CHECK_CODE(code, lino, _end);
 
       if (pContext->recovering && recoveryDone) {
         ST_TASK_DLOG("stop fetch wal metas since recovery is done, pool size: %" PRId64, pContext->metaPool.size);
@@ -6384,6 +6433,9 @@ static int32_t stRealtimeContextProcPullRsp(SSTriggerRealtimeContext *pContext,
         if (latestVersionTime != INT64_MAX) {
           atomic_store_64(&pTask->latestVersionTime, latestVersionTime);
         }
+        // Check for idle groups
+        code = stRealtimeContextCheckIdleGroup(pContext);
+        QUERY_CHECK_CODE(code, lino, _end);
       }
 
       pContext->catchUp = (TD_DLIST_NELES(&pContext->groupsToCheck) == 0);
@@ -9001,7 +9053,6 @@ static void stRealtimeGroupClearTempState(SSTriggerRealtimeGroup *pGroup) {
 
   pContext->needPseudoCols = false;
   pContext->needMergeWindow = false;
-  pContext->needCheckAgain = false;
   if (pContext->pSorter != NULL) {
     stNewTimestampSorterReset(pContext->pSorter);
   }
@@ -9100,7 +9151,7 @@ static int32_t stRealtimeGroupAddMeta(SSTriggerRealtimeGroup *pGroup, int32_t vg
   SObjList                 *pMetas = NULL;
 
   // Update idle trigger timestamps when receiving data
-  if (pTask->idleTimeoutMs > 0) {
+  if (pTask->idleTimeoutMs > 0 && !pContext->recovering) {
     int64_t prevRecvTimeMono = pGroup->lastRecvTimeMono;
     int64_t prevRecvTimeWall = pGroup->lastRecvTimeWall;
     pGroup->lastRecvTimeMono = taosGetMonoTimestampMs();
@@ -9425,10 +9476,9 @@ static int32_t stRealtimeGroupDoSlidingCheck(SSTriggerRealtimeGroup *pGroup) {
     while (newWin.range.skey <= pGroup->newThreshold) {
       void *px = taosArrayPush(pContext->pWindows, &newWin);
       QUERY_CHECK_NULL(px, code, lino, _end, terrno);
-      if (pContext->walMode == STRIGGER_WAL_META_ONLY &&
-          TARRAY_SIZE(pContext->pWindows) >= STREAM_CALC_REQ_MAX_WIN_NUM) {
+      if (TARRAY_SIZE(pContext->pWindows) >= STREAM_CALC_REQ_MAX_WIN_NUM) {
         pContext->needCheckAgain = true;
-        goto _end;
+        break;
       }
       stTriggerTaskNextTimeWindow(pTask, &newWin.range);
     }
diff --git a/source/libs/scheduler/src/schTask.c b/source/libs/scheduler/src/schTask.c
@@ -1097,18 +1097,29 @@ int32_t schLaunchRemoteTask(SSchJob *pJob, SSchTask *pTask) {
   if (NULL == pTask->msg) {  // TODO add more detailed reason for failure
     SCH_LOCK(SCH_WRITE, &pTask->planLock);
     code = qSubPlanToMsg(plan, &pTask->msg, &pTask->msgLen);
+    if (TSDB_CODE_SUCCESS == code && tsQueryPlannerTrace) {
+      if (SUBPLAN_TYPE_MODIFY == plan->subplanType) {
+        SDataInserterNode *insert = (SDataInserterNode *)plan->pDataSink;
+        SCH_TASK_DLOG("MODIFY plan, tables:%d, payload size:%u", insert ? insert->numOfTables : 0,
+                      insert ? insert->size : 0);
+      } else {
+        char   *msg = NULL;
+        int32_t msgLen = 0;
+        int32_t traceCode = qSubPlanToString(plan, &msg, &msgLen);
+        if (TSDB_CODE_SUCCESS == traceCode) {
+          SCH_TASK_DLOGL("physical plan len:%d, %s", msgLen, msg);
+          taosMemoryFree(msg);
+        } else {
+          SCH_TASK_WLOG("plan trace failed, code:%s", tstrerror(traceCode));
+        }
+      }
+    }
     SCH_UNLOCK(SCH_WRITE, &pTask->planLock);
 
     if (TSDB_CODE_SUCCESS != code) {
       SCH_TASK_ELOG("failed to create physical plan, code:%s, msg:%p, len:%d", tstrerror(code), pTask->msg,
                     pTask->msgLen);
       SCH_ERR_RET(code);
-    } else if (tsQueryPlannerTrace) {
-      char   *msg = NULL;
-      int32_t msgLen = 0;
-      SCH_ERR_RET(qSubPlanToString(plan, &msg, &msgLen));
-      SCH_TASK_DLOGL("physical plan len:%d, %s", msgLen, msg);
-      taosMemoryFree(msg);
     }
   }
 
diff --git a/source/os/src/osSysinfo.c b/source/os/src/osSysinfo.c
diff --git a/test/cases/18-StreamProcessing/03-TriggerMode/test_sliding_trigger_check_again.py b/test/cases/18-StreamProcessing/03-TriggerMode/test_sliding_trigger_check_again.py
diff --git a/test/ci/cases.task b/test/ci/cases.task

Original file line number	Diff line number	Diff line change
`@@ -75,11 +75,13 @@ void mstDestroySStmTaskToDeployExt(void* param) {`
`75`	`75`	`case STREAM_READER_TASK:`
`76`	`76`	`if (!pExt->deploy.msg.reader.triggerReader) {`
`77`	`77`	`SStreamReaderDeployFromCalc* pCalcReaderDeploy = &pExt->deploy.msg.reader.msg.calc;`
`78`		`- taosMemoryFreeClear(pCalcReaderDeploy->calcScanPlan);`
	`78`	`+ if (pCalcReaderDeploy->freeScanPlan) {`
	`79`	`+ taosMemoryFreeClear(pCalcReaderDeploy->calcScanPlan);`
	`80`	`+ }`
`79`	`81`	`}`
`80`	`82`	`break;`
`81`	`83`	`default:`
`82`		`- break;;`
	`84`	`+ break;`
`83`	`85`	`}`
`84`	`86`	`}`
`85`	`87`