add delete workflow instance when delete process definition (#8131)

Co-authored-by: Ncaishunfeng <534328519@qq.com>

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java

@@ -709,10 +709,43 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
         if (deleteRelation == 0) {
             logger.warn("The process definition has not relation, it will be delete successfully");
         }
+
+        try {
+            syncDeleteWorkflowInstanceByCode(processDefinition.getCode());
+        } catch (Exception e) {
+            logger.error("delete workflow instance error", e);
+        }
+
         putMsg(result, Status.SUCCESS);
         return result;
     }
 
+    /**
+     * delete workflow instance by processDefinitionCode
+     * 1.delete processInstances
+     * 2.delete subWorkProcesses
+     * 3.delete processMap
+     * 4.delete taskInstances
+     *
+     * todo delete syncly may take a long time when many processInstance
+     * @param processDefinitionCode
+     */
+    private void syncDeleteWorkflowInstanceByCode(long processDefinitionCode) {
+        int pageSize = 100;
+        while (true) {
+            List<ProcessInstance> deleteProcessInstances = processInstanceService.queryByProcessDefineCode(processDefinitionCode, pageSize);
+            if (CollectionUtils.isEmpty(deleteProcessInstances)) {
+                break;
+            }
+            for (ProcessInstance deleteProcessInstance : deleteProcessInstances) {
+                processService.deleteWorkProcessInstanceById(deleteProcessInstance.getId());
+                processService.deleteAllSubWorkProcessByParentId(deleteProcessInstance.getId());
+                processService.deleteWorkProcessMapByParentId(deleteProcessInstance.getId());
+                processService.deleteWorkTaskInstanceByProcessInstanceId(deleteProcessInstance.getId());
+            }
+        }
+    }
+
     /**
      * release process definition: online / offline
      *

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java

@@ -17,7 +17,6 @@
 
 package org.apache.dolphinscheduler.service.process;
 
-import static java.util.stream.Collectors.toSet;
 import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
 import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_EMPTY_SUB_PROCESS;
@@ -28,6 +27,8 @@ import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID;
 import static org.apache.dolphinscheduler.common.Constants.LOCAL_PARAMS;
 
+import static java.util.stream.Collectors.toSet;
+
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.AuthorizationType;
 import org.apache.dolphinscheduler.common.enums.CommandType;