scan op state

source/libs/executor/src/scanoperator.c

@@ -1775,6 +1775,7 @@ void streamScanOperatorSaveCheckpoint(SStreamScanInfo* pInfo) {
   int32_t len = streamScanOperatorEncode(pInfo, &pBuf);
   pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_SCAN_OP_CHECKPOINT_NAME, strlen(STREAM_SCAN_OP_CHECKPOINT_NAME), pBuf, len);
   taosMemoryFree(pBuf);
+  pInfo->stateStore.streamStateCommit(pInfo->pState);
 }
 
 // other properties are recovered from the execution plan
@@ -2188,7 +2189,6 @@ FETCH_NEXT_BLOCK:
 
     if (pBlock->info.type == STREAM_CHECKPOINT) {
       streamScanOperatorSaveCheckpoint(pInfo);
-      pAPI->stateStore.streamStateCommit(pInfo->pState);
     }
     printDataBlock(pBlock, "stream scan ck");
     return pInfo->pCheckpointRes;