Merge pull request #23946 from taosdata/fix/snodeBackendCrash

Fix/snode backend crash

include/libs/executor/storageapi.h

@@ -102,11 +102,11 @@ typedef struct SMTbCursor {
 } SMTbCursor;
 
 typedef struct SMCtbCursor {
-  SMeta   *pMeta;
+  SMeta*   pMeta;
-  void    *pCur;
+  void*    pCur;
   tb_uid_t suid;
-  void    *pKey;
+  void*    pKey;
-  void    *pVal;
+  void*    pVal;
   int      kLen;
   int      vLen;
   int8_t   paused;
@@ -263,21 +263,22 @@ typedef struct SStoreMeta {
   void* (*storeGetIndexInfo)();
   void* (*getInvertIndex)(void* pVnode);
   // support filter and non-filter cases. [vnodeGetCtbIdList & vnodeGetCtbIdListByFilter]
-  int32_t (*getChildTableList)( void* pVnode, int64_t suid, SArray* list);
+  int32_t (*getChildTableList)(void* pVnode, int64_t suid, SArray* list);
   int32_t (*storeGetTableList)(void* pVnode, int8_t type, SArray* pList);
   void* storeGetVersionRange;
   void* storeGetLastTimestamp;
 
   int32_t (*getTableSchema)(void* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid);  // tsdbGetTableSchema
-  int32_t (*getNumOfChildTables)( void* pVnode, int64_t uid, int64_t* numOfTables, int32_t* numOfCols);
+  int32_t (*getNumOfChildTables)(void* pVnode, int64_t uid, int64_t* numOfTables, int32_t* numOfCols);
-  void (*getBasicInfo)(void* pVnode, const char** dbname, int32_t* vgId, int64_t* numOfTables, int64_t* numOfNormalTables);
+  void (*getBasicInfo)(void* pVnode, const char** dbname, int32_t* vgId, int64_t* numOfTables,
+                       int64_t* numOfNormalTables);
 
   int64_t (*getNumOfRowsInMem)(void* pVnode);
 
-  SMCtbCursor* (*openCtbCursor)(void *pVnode, tb_uid_t uid, int lock);
+  SMCtbCursor* (*openCtbCursor)(void* pVnode, tb_uid_t uid, int lock);
   int32_t (*resumeCtbCursor)(SMCtbCursor* pCtbCur, int8_t first);
   void (*pauseCtbCursor)(SMCtbCursor* pCtbCur);
-  void         (*closeCtbCursor)(SMCtbCursor *pCtbCur);
+  void (*closeCtbCursor)(SMCtbCursor* pCtbCur);
   tb_uid_t (*ctbCursorNext)(SMCtbCursor* pCur);
 } SStoreMeta;
 
@@ -366,11 +367,11 @@ typedef struct SStateStore {
   TSKEY (*updateInfoFillBlockData)(SUpdateInfo* pInfo, SSDataBlock* pBlock, int32_t primaryTsCol);
   bool (*updateInfoIsUpdated)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts);
   bool (*updateInfoIsTableInserted)(SUpdateInfo* pInfo, int64_t tbUid);
-  bool         (*isIncrementalTimeStamp)(SUpdateInfo *pInfo, uint64_t tableId, TSKEY ts);
+  bool (*isIncrementalTimeStamp)(SUpdateInfo* pInfo, uint64_t tableId, TSKEY ts);
 
   void (*updateInfoDestroy)(SUpdateInfo* pInfo);
-  void (*windowSBfDelete)(SUpdateInfo *pInfo, uint64_t count);
+  void (*windowSBfDelete)(SUpdateInfo* pInfo, uint64_t count);
-  void (*windowSBfAdd)(SUpdateInfo *pInfo, uint64_t count);
+  void (*windowSBfAdd)(SUpdateInfo* pInfo, uint64_t count);
 
   SUpdateInfo* (*updateInfoInitP)(SInterval* pInterval, int64_t watermark, bool igUp);
   void (*updateInfoAddCloseWindowSBF)(SUpdateInfo* pInfo);
@@ -397,6 +398,7 @@ typedef struct SStateStore {
   void (*streamStateDestroy)(SStreamState* pState, bool remove);
   int32_t (*streamStateDeleteCheckPoint)(SStreamState* pState, TSKEY mark);
   void (*streamStateReloadInfo)(SStreamState* pState, TSKEY ts);
+  void (*streamStateCopyBackend)(SStreamState* src, SStreamState* dst);
 } SStateStore;
 
 typedef struct SStorageAPI {

include/libs/function/function.h

@@ -171,6 +171,7 @@ typedef struct {
   int32_t                  taskId;
   int64_t                  streamId;
   int64_t                  streamBackendRid;
+  int8_t                   dump;
 } SStreamState;
 
 typedef struct SFunctionStateStore {

include/libs/stream/streamState.h

@@ -50,7 +50,7 @@ void    streamStateSetNumber(SStreamState* pState, int32_t number);
 int32_t streamStateSaveInfo(SStreamState* pState, void* pKey, int32_t keyLen, void* pVal, int32_t vLen);
 int32_t streamStateGetInfo(SStreamState* pState, void* pKey, int32_t keyLen, void** pVal, int32_t* pLen);
 
-//session window
+// session window
 int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal, int32_t* pVLen);
 int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, void* value, int32_t vLen);
 int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen);
@@ -65,7 +65,7 @@ SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSess
 SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, const SSessionKey* key);
 SStreamStateCur* streamStateSessionSeekKeyCurrentNext(SStreamState* pState, const SSessionKey* key);
 
-//state window
+// state window
 int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
                                       state_key_cmpr_fn fn, void** pVal, int32_t* pVLen);
 
@@ -96,6 +96,9 @@ int32_t streamStatePutParName(SStreamState* pState, int64_t groupId, const char*
 int32_t streamStateGetParName(SStreamState* pState, int64_t groupId, void** pVal);
 
 void streamStateReloadInfo(SStreamState* pState, TSKEY ts);
+
+void streamStateCopyBackend(SStreamState* src, SStreamState* dst);
+
 SStreamStateCur* createStreamStateCursor();
 
 /***compare func **/

source/dnode/snode/src/snode.c

@@ -67,6 +67,7 @@ int32_t sndExpandTask(SSnode *pSnode, SStreamTask *pTask, int64_t nextProcessVer
     sndDebug("s-task:%s state:%p", pTask->id.idStr, pTask->pState);
   }
   
+
   int32_t     numOfVgroups = (int32_t)taosArrayGetSize(pTask->upstreamInfo.pList);
   SReadHandle handle = {
       .checkpointId = pTask->chkInfo.checkpointId,

source/dnode/snode/src/snodeInitApi.c

@@ -14,8 +14,8 @@
  */
 
 #include "storageapi.h"
-#include "tstreamUpdate.h"
 #include "streamState.h"
+#include "tstreamUpdate.h"
 
 static void initStateStoreAPI(SStateStore* pStore);
 static void initFunctionStateStore(SFunctionStateStore* pStore);
@@ -100,9 +100,10 @@ void initStateStoreAPI(SStateStore* pStore) {
   pStore->streamStateClose = streamStateClose;
   pStore->streamStateBegin = streamStateBegin;
   pStore->streamStateCommit = streamStateCommit;
-  pStore->streamStateDestroy= streamStateDestroy;
+  pStore->streamStateDestroy = streamStateDestroy;
   pStore->streamStateDeleteCheckPoint = streamStateDeleteCheckPoint;
   pStore->streamStateReloadInfo = streamStateReloadInfo;
+  pStore->streamStateCopyBackend = streamStateCopyBackend;
 }
 
 void initFunctionStateStore(SFunctionStateStore* pStore) {

source/dnode/vnode/src/tqCommon/tqCommon.c

@@ -13,9 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "tstream.h"
 #include "tmsgcb.h"
 #include "tq.h"
+#include "tstream.h"
 
 typedef struct STaskUpdateEntry {
   int64_t streamId;
@@ -42,7 +42,7 @@ int32_t tqStreamTaskStartAsync(SStreamMeta* pMeta, SMsgCb* cb, bool restart) {
   tqDebug("vgId:%d start all %d stream task(s) async", vgId, numOfTasks);
   pRunReq->head.vgId = vgId;
   pRunReq->streamId = 0;
-  pRunReq->taskId = restart? STREAM_EXEC_RESTART_ALL_TASKS_ID:STREAM_EXEC_START_ALL_TASKS_ID;
+  pRunReq->taskId = restart ? STREAM_EXEC_RESTART_ALL_TASKS_ID : STREAM_EXEC_START_ALL_TASKS_ID;
 
   SRpcMsg msg = {.msgType = TDMT_STREAM_TASK_RUN, .pCont = pRunReq, .contLen = sizeof(SStreamTaskRunReq)};
   tmsgPutToQueue(cb, STREAM_QUEUE, &msg);
@@ -166,7 +166,8 @@ int32_t tqStreamTaskProcessUpdateReq(SStreamMeta* pMeta, SMsgCb* cb, SRpcMsg* pM
     streamMetaWUnLock(pMeta);
   } else {
     if (!restored) {
-      tqDebug("vgId:%d vnode restore not completed, not restart the tasks, clear the start after nodeUpdate flag", vgId);
+      tqDebug("vgId:%d vnode restore not completed, not restart the tasks, clear the start after nodeUpdate flag",
+              vgId);
       pMeta->startInfo.tasksWillRestart = 0;
       streamMetaWUnLock(pMeta);
     } else {
@@ -251,7 +252,7 @@ int32_t tqStreamTaskProcessDispatchReq(SStreamMeta* pMeta, SRpcMsg* pMsg) {
   SStreamTask* pTask = streamMetaAcquireTask(pMeta, req.streamId, req.taskId);
   if (pTask) {
     SRpcMsg rsp = {.info = pMsg->info, .code = 0};
-    if (streamProcessDispatchMsg(pTask, &req, &rsp) != 0){
+    if (streamProcessDispatchMsg(pTask, &req, &rsp) != 0) {
       return -1;
     }
     tDeleteStreamDispatchReq(&req);
@@ -355,8 +356,8 @@ int32_t tqStreamTaskProcessScanHistoryFinishReq(SStreamMeta* pMeta, SRpcMsg* pMs
 
   SStreamTask* pTask = streamMetaAcquireTask(pMeta, req.streamId, req.downstreamTaskId);
   if (pTask == NULL) {
-    tqError("vgId:%d process scan history finish msg, failed to find task:0x%x, it may be destroyed",
+    tqError("vgId:%d process scan history finish msg, failed to find task:0x%x, it may be destroyed", pMeta->vgId,
-            pMeta->vgId, req.downstreamTaskId);
+            req.downstreamTaskId);
     return -1;
   }
 
@@ -381,8 +382,8 @@ int32_t tqStreamTaskProcessScanHistoryFinishRsp(SStreamMeta* pMeta, SRpcMsg* pMs
 
   SStreamTask* pTask = streamMetaAcquireTask(pMeta, req.streamId, req.upstreamTaskId);
   if (pTask == NULL) {
-    tqError("vgId:%d process scan history finish rsp, failed to find task:0x%x, it may be destroyed",
+    tqError("vgId:%d process scan history finish rsp, failed to find task:0x%x, it may be destroyed", pMeta->vgId,
-            pMeta->vgId, req.upstreamTaskId);
+            req.upstreamTaskId);
     return -1;
   }
 
@@ -428,7 +429,8 @@ int32_t tqStreamTaskProcessCheckReq(SStreamMeta* pMeta, SRpcMsg* pMsg) {
 
   // only the leader node handle the check request
   if (pMeta->role == NODE_ROLE_FOLLOWER) {
-    tqError("s-task:0x%x invalid check msg from upstream:0x%x(vgId:%d), vgId:%d is follower, not handle check status msg",
+    tqError(
+        "s-task:0x%x invalid check msg from upstream:0x%x(vgId:%d), vgId:%d is follower, not handle check status msg",
         taskId, req.upstreamTaskId, req.upstreamNodeId, pMeta->vgId);
     rsp.status = TASK_DOWNSTREAM_NOT_LEADER;
   } else {
@@ -439,7 +441,8 @@ int32_t tqStreamTaskProcessCheckReq(SStreamMeta* pMeta, SRpcMsg* pMsg) {
 
       char* p = NULL;
       streamTaskGetStatus(pTask, &p);
-      tqDebug("s-task:%s status:%s, stage:%"PRId64" recv task check req(reqId:0x%" PRIx64 ") task:0x%x (vgId:%d), check_status:%d",
+      tqDebug("s-task:%s status:%s, stage:%" PRId64 " recv task check req(reqId:0x%" PRIx64
+              ") task:0x%x (vgId:%d), check_status:%d",
               pTask->id.idStr, p, rsp.oldStage, rsp.reqId, rsp.upstreamTaskId, rsp.upstreamNodeId, rsp.status);
     } else {
       rsp.status = TASK_DOWNSTREAM_NOT_READY;
@@ -526,7 +529,8 @@ int32_t tqStreamTaskProcessCheckpointReadyMsg(SStreamMeta* pMeta, SRpcMsg* pMsg)
   return code;
 }
 
-int32_t tqStreamTaskProcessDeployReq(SStreamMeta* pMeta, int64_t sversion, char* msg, int32_t msgLen, bool isLeader, bool restored) {
+int32_t tqStreamTaskProcessDeployReq(SStreamMeta* pMeta, int64_t sversion, char* msg, int32_t msgLen, bool isLeader,
+                                     bool restored) {
   int32_t code = 0;
   int32_t vgId = pMeta->vgId;
 
@@ -566,7 +570,8 @@ int32_t tqStreamTaskProcessDeployReq(SStreamMeta* pMeta, int64_t sversion, char*
   streamMetaWUnLock(pMeta);
 
   if (code < 0) {
-    tqError("failed to add s-task:0x%x into vgId:%d meta, total:%d, code:%s", vgId, taskId, numOfTasks, tstrerror(code));
+    tqError("failed to add s-task:0x%x into vgId:%d meta, total:%d, code:%s", vgId, taskId, numOfTasks,
+            tstrerror(code));
     tFreeStreamTask(pTask);
     return code;
   }
@@ -716,7 +721,7 @@ static int32_t restartStreamTasks(SStreamMeta* pMeta, bool isLeader) {
   int32_t code = 0;
   int64_t st = taosGetTimestampMs();
 
-  while(1) {
+  while (1) {
     int32_t startVal = atomic_val_compare_exchange_32(&pMeta->startInfo.taskStarting, 0, 1);
     if (startVal == 0) {
       break;
@@ -739,7 +744,7 @@ static int32_t restartStreamTasks(SStreamMeta* pMeta, bool isLeader) {
   streamMetaClear(pMeta);
 
   int64_t el = taosGetTimestampMs() - st;
-  tqInfo("vgId:%d close&reload state elapsed time:%.3fs", vgId, el/1000.);
+  tqInfo("vgId:%d close&reload state elapsed time:%.3fs", vgId, el / 1000.);
 
   code = streamMetaLoadAllTasks(pMeta);
   if (code != TSDB_CODE_SUCCESS) {
@@ -800,5 +805,3 @@ int32_t tqStreamTaskProcessRunReq(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLead
     return -1;
   }
 }
-
-

source/dnode/vnode/src/vnd/vnodeInitApi.c

@@ -215,6 +215,7 @@ void initStateStoreAPI(SStateStore* pStore) {
   pStore->streamStateDestroy = streamStateDestroy;
   pStore->streamStateDeleteCheckPoint = streamStateDeleteCheckPoint;
   pStore->streamStateReloadInfo = streamStateReloadInfo;
+  pStore->streamStateCopyBackend = streamStateCopyBackend;
 }
 
 void initMetaReaderAPI(SStoreMetaReader* pMetaReader) {

source/libs/executor/src/streamtimewindowoperator.c

@@ -408,6 +408,11 @@ void destroyStreamFinalIntervalOperatorInfo(void* param) {
   taosArrayDestroy(pInfo->pDelWins);
   blockDataDestroy(pInfo->pDelRes);
   pInfo->stateStore.streamFileStateDestroy(pInfo->pState->pFileState);
+
+  if (pInfo->pState->dump == 1) {
+    taosMemoryFreeClear(pInfo->pState->pTdbState->pOwner);
+    taosMemoryFreeClear(pInfo->pState->pTdbState);
+  }
   taosMemoryFreeClear(pInfo->pState);
 
   nodesDestroyNode((SNode*)pInfo->pPhyNode);
@@ -1462,7 +1467,11 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
   initBasicInfo(&pInfo->binfo, pResBlock);
 
   pInfo->pState = taosMemoryCalloc(1, sizeof(SStreamState));
-  *(pInfo->pState) = *(pTaskInfo->streamInfo.pState);
+  qInfo("open state %p", pInfo->pState);
+  pAPI->stateStore.streamStateCopyBackend(pTaskInfo->streamInfo.pState, pInfo->pState);
+  //*(pInfo->pState) = *(pTaskInfo->streamInfo.pState);
+
+  qInfo("copy state %p to %p", pTaskInfo->streamInfo.pState, pInfo->pState);
 
   pAPI->stateStore.streamStateSetNumber(pInfo->pState, -1);
   int32_t code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str,
@@ -3338,7 +3347,8 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
   SColumnInfoData* pKeyColInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->stateCol.slotId);
   for (int32_t i = 0; i < rows; i += winRows) {
     if (pInfo->ignoreExpiredData && checkExpiredData(&pInfo->streamAggSup.stateStore, pInfo->streamAggSup.pUpdateInfo,
-                                                     &pInfo->twAggSup, pSDataBlock->info.id.uid, tsCols[i]) || colDataIsNull_s(pKeyColInfo, i)) {
+                                                     &pInfo->twAggSup, pSDataBlock->info.id.uid, tsCols[i]) ||
+        colDataIsNull_s(pKeyColInfo, i)) {
       i++;
       continue;
     }

source/libs/stream/src/streamState.c

@@ -91,8 +91,8 @@ int stateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
 }
 
 SStreamState* streamStateOpen(char* path, void* pTask, bool specPath, int32_t szPage, int32_t pages) {
-  stDebug("open stream state, %s", path);
   SStreamState* pState = taosMemoryCalloc(1, sizeof(SStreamState));
+  stDebug("open stream state %p, %s", pState, path);
   if (pState == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
@@ -710,7 +710,8 @@ int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, void
 #endif
 }
 
-int32_t streamStateSessionAllocWinBuffByNextPosition(SStreamState* pState, SStreamStateCur* pCur, const SSessionKey* pKey, void** pVal, int32_t* pVLen) {
+int32_t streamStateSessionAllocWinBuffByNextPosition(SStreamState* pState, SStreamStateCur* pCur,
+                                                     const SSessionKey* pKey, void** pVal, int32_t* pVLen) {
 #ifdef USE_ROCKSDB
   return allocSessioncWinBuffByNextPosition(pState->pFileState, pCur, pKey, pVal, pVLen);
 #else
@@ -1115,6 +1116,20 @@ int32_t streamStateDeleteCheckPoint(SStreamState* pState, TSKEY mark) {
 
 void streamStateReloadInfo(SStreamState* pState, TSKEY ts) { streamFileStateReloadInfo(pState->pFileState, ts); }
 
+void streamStateCopyBackend(SStreamState* src, SStreamState* dst) {
+  dst->pFileState = src->pFileState;
+  dst->parNameMap = src->parNameMap;
+  dst->number = src->number;
+  dst->taskId = src->taskId;
+  dst->streamId = src->streamId;
+  if (dst->pTdbState == NULL) {
+    dst->pTdbState = taosMemoryCalloc(1, sizeof(STdbState));
+    dst->pTdbState->pOwner = taosMemoryCalloc(1, sizeof(SStreamTask));
+  }
+  dst->dump = 1;
+  dst->pTdbState->pOwner->pBackend = src->pTdbState->pOwner->pBackend;
+  return;
+}
 SStreamStateCur* createStreamStateCursor() {
   SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
   pCur->buffIndex = -1;

tests/parallel_test/cases.task

@@ -26,7 +26,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 8-stream/pause_resume_test.py
 #,,n,system-test,python3 ./test.py -f 8-stream/vnode_restart.py -N 4
 #,,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
+,,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/tbname_vgroup.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stbJoin.py