perf(flow): optimize performance for flow instances query (#1450)

* pick from 4.2.x

* optimize query

* fix

Author: ungreat

server/integration-test/src/test/java/com/oceanbase/odc/service/flow/FlowInstanceServiceTest.java

@@ -61,6 +61,7 @@
 import com.oceanbase.odc.core.shared.exception.OverLimitException;
 import com.oceanbase.odc.metadb.flow.FlowInstanceEntity;
 import com.oceanbase.odc.metadb.flow.FlowInstanceRepository;
+import com.oceanbase.odc.metadb.flow.FlowInstanceRepository.ParentInstanceIdCount;
 import com.oceanbase.odc.metadb.flow.GateWayInstanceRepository;
 import com.oceanbase.odc.metadb.flow.NodeInstanceEntityRepository;
 import com.oceanbase.odc.metadb.flow.SequenceInstanceRepository;
@@ -443,6 +444,14 @@ public void listStatus() {
         Assert.assertEquals(0, status.size());
     }
 
+    @Test
+    public void testFindByParentInstanceIdIn() {
+        createChildFlowInstance("test", 1L);
+        List<ParentInstanceIdCount> byParentInstanceIdIn = flowInstanceRepository.findByParentInstanceIdIn(
+                Arrays.asList(1L, 2L));
+        Assert.assertEquals(byParentInstanceIdIn.size(), 1);
+    }
+
     private void buildFlowInstance(FlowInstance flowInstance) {
         buildFlowInstanceWithTaskType(flowInstance, TaskType.ASYNC);
     }
@@ -486,6 +495,10 @@ private FlowInstance createFlowInstance(String name) {
         return flowFactory.generateFlowInstance(name, null, 1L, null);
     }
 
+    private FlowInstance createChildFlowInstance(String name, Long parentFloweInstanceId) {
+        return flowFactory.generateFlowInstance(name, parentFloweInstanceId, 1L, null);
+    }
+
     private FlowTaskInstance createTaskInstance(Long flowInstanceId, TaskEntity taskEntity,
             ExecutionStrategyConfig config) {
         FlowTaskInstance taskInstance = new FlowTaskInstance(taskEntity.getTaskType(),

server/odc-migrate/src/main/resources/migrate/common/V_4_2_3_15__add_flow_instance_index.sql

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2024 OceanBase.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+alter table `flow_instance` add index `flow_instance_parent_ins_id`(`parent_instance_id`);
+
+alter table `flow_instance_node_task` add index `flow_instance_node_task_flow_instance_id`(`flow_instance_id`);
+
+alter table `flow_instance` add index `flow_instance_parent_oid_ct_id`(`organization_id`,`create_time`,`id`);
+
+alter table `flow_instance_node` add index `flow_instance_node_intsance_id`(`instance_id`);
+
+alter table `flow_instance` add index `flow_instance_process_definition_id`(`process_definition_id`);
+
+CREATE or replace  VIEW `list_flow_instance_view` AS
+select
+    `flow_instance`.`id` AS `id`,
+    `flow_instance`.`create_time` AS `create_time`,
+    `flow_instance`.`update_time` AS `update_time`,
+    `flow_instance`.`name` AS `name`,
+    `flow_instance`.`flow_config_id` AS `flow_config_id`,
+    `flow_instance`.`creator_id` AS `creator_id`,
+    `flow_instance`.`organization_id` AS `organization_id`,
+    `flow_instance`.`process_definition_id` AS `process_definition_id`,
+    `flow_instance`.`process_instance_id` AS `process_instance_id`,
+    `flow_instance`.`status` AS `status`,
+    `flow_instance`.`flow_config_snapshot_xml` AS `flow_config_snapshot_xml`,
+    `flow_instance`.`description` AS `description`,
+    `flow_instance`.`parent_instance_id` AS `parent_instance_id`,
+    `flow_instance`.`project_id` AS `project_id`,
+    `flow_instance_node_task`.`task_type` AS `task_type`
+from
+    (
+        `flow_instance` join `flow_instance_node_task` on  ( `flow_instance`.`id` = `flow_instance_node_task`.`flow_instance_id`)
+    )
+group by `flow_instance`.`id`,`flow_instance_node_task`.`task_type`

server/odc-service/src/main/java/com/oceanbase/odc/metadb/flow/FlowInstanceRepository.java

@@ -21,12 +21,12 @@
 
 import javax.transaction.Transactional;
 
-import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
 
+import com.oceanbase.odc.config.jpa.OdcJpaRepository;
 import com.oceanbase.odc.core.shared.constant.FlowStatus;
 import com.oceanbase.odc.core.shared.constant.TaskType;
 
@@ -39,7 +39,7 @@
  * @see org.springframework.data.jpa.repository.JpaRepository
  */
 public interface FlowInstanceRepository
-        extends JpaRepository<FlowInstanceEntity, Long>, JpaSpecificationExecutor<FlowInstanceEntity> {
+        extends OdcJpaRepository<FlowInstanceEntity, Long>, JpaSpecificationExecutor<FlowInstanceEntity> {
 
     List<FlowInstanceEntity> findByIdIn(Collection<Long> ids);
 
@@ -82,6 +82,7 @@ int updateProcessDefinitionIdById(@Param("flowInstanceId") Long flowInstanceId,
 
     List<FlowInstanceEntity> findByParentInstanceId(Long parentInstanceId);
 
+
     @Query(value = "select a.parent_instance_id from flow_instance a left join flow_instance_node_task b on a.id = b.flow_instance_id"
             + " where a.id=:id and task_task_id is not null and b.task_type='ALTER_SCHEDULE' LIMIT 1",
             nativeQuery = true)
@@ -99,4 +100,14 @@ List<FlowInstanceEntity> findByFlowInstanceIdsAndTaskType(@Param("ids") Collecti
             nativeQuery = true)
     Set<FlowInstanceEntity> findByScheduleIdAndStatus(@Param("scheduleIds") Set<Long> scheduleIds,
             @Param("status") FlowStatus status);
+
+    @Query("select e.parentInstanceId as parentInstanceId, count(1) as count from FlowInstanceEntity e where e.parentInstanceId in (?1) group by parentInstanceId")
+    List<ParentInstanceIdCount> findByParentInstanceIdIn(Collection<Long> parentInstanceId);
+
+    interface ParentInstanceIdCount {
+
+        Long getParentInstanceId();
+
+        Integer getCount();
+    }
 }

server/odc-service/src/main/java/com/oceanbase/odc/metadb/flow/ServiceTaskInstanceRepository.java

@@ -15,6 +15,7 @@
  */
 package com.oceanbase.odc.metadb.flow;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
@@ -78,9 +79,10 @@ Optional<ServiceTaskInstanceEntity> findByInstanceTypeAndName(@Param("instanceTy
 
     @Query(value = "select b.* from flow_instance a left join flow_instance_node_task b "
             + "on a.id = b.flow_instance_id "
-            + "where a.parent_instance_id=:id and b.task_type=:#{#type.name()}", nativeQuery = true)
-    List<ServiceTaskInstanceEntity> findByScheduleIdAndTaskType(@Param("id") Long scheduleId,
-            @Param("type") TaskType type);
+            + "where a.parent_instance_id=:id and b.task_type=:#{#type.name()} and b.status in (:statuses)",
+            nativeQuery = true)
+    List<ServiceTaskInstanceEntity> findByScheduleIdAndTaskTypeAndStatusIn(@Param("id") Long scheduleId,
+            @Param("type") TaskType type, @Param("statuses") Collection<?> statuses);
 
     default List<ServiceTaskInstanceEntity> batchCreate(List<ServiceTaskInstanceEntity> entities) {
         String sql = InsertSqlTemplateBuilder.from("flow_instance_node_task")

server/odc-service/src/main/java/com/oceanbase/odc/service/flow/factory/FlowResponseMapperFactory.java

@@ -35,12 +35,14 @@
 import org.springframework.data.jpa.domain.Specification;
 import org.springframework.stereotype.Component;
 
+import com.google.common.base.MoreObjects;
 import com.oceanbase.odc.common.util.StringUtils;
 import com.oceanbase.odc.metadb.connection.ConnectionConfigRepository;
 import com.oceanbase.odc.metadb.connection.ConnectionEntity;
 import com.oceanbase.odc.metadb.connection.ConnectionSpecs;
 import com.oceanbase.odc.metadb.flow.FlowInstanceEntity;
 import com.oceanbase.odc.metadb.flow.FlowInstanceRepository;
+import com.oceanbase.odc.metadb.flow.FlowInstanceRepository.ParentInstanceIdCount;
 import com.oceanbase.odc.metadb.flow.ServiceTaskInstanceEntity;
 import com.oceanbase.odc.metadb.flow.ServiceTaskInstanceRepository;
 import com.oceanbase.odc.metadb.flow.ServiceTaskInstanceSpecs;
@@ -271,8 +273,14 @@ private FlowInstanceMapper generateMapper(@NonNull Collection<Long> flowInstance
                 .collect(Collectors.groupingBy(ServiceTaskInstanceEntity::getFlowInstanceId,
                         Collectors.mapping(ServiceTaskInstanceEntity::getStrategy, Collectors.toList())));
 
+        Map<Long, Integer> parentInstanceIdMap = flowInstanceRepository
+                .findByParentInstanceIdIn(flowInstanceIds)
+                .stream().collect(
+                        Collectors.toMap(ParentInstanceIdCount::getParentInstanceId, ParentInstanceIdCount::getCount));
+
         Map<Long, Boolean> flowInstanceId2Rollbackable = flowInstanceIds.stream().collect(Collectors
-                .toMap(Function.identity(), id -> flowInstanceRepository.findByParentInstanceId(id).size() == 0));
+                .toMap(Function.identity(),
+                        id -> MoreObjects.firstNonNull(parentInstanceIdMap.get(id), 0) == 0));
 
         /**
          * In order to improve the interface efficiency, it is necessary to find out the task entity

server/odc-service/src/main/java/com/oceanbase/odc/service/flow/model/FlowNodeStatus.java

@@ -16,12 +16,12 @@
 package com.oceanbase.odc.service.flow.model;
 
 import java.util.Arrays;
-import java.util.HashSet;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import com.oceanbase.odc.service.flow.instance.FlowTaskInstance;
 
-import lombok.NonNull;
+import lombok.Getter;
 
 /**
  * Process state, used to mark the execution state of process instance nodes, including process
@@ -31,58 +31,62 @@
  * @date 2022-02-07 11:15
  * @since ODC_release_3.3.0
  */
+@Getter
 public enum FlowNodeStatus {
     /**
      * Process node has been created but has not started execution
      */
-    CREATED,
+    CREATED(false, false),
     /**
      * Process node is executing
      */
-    EXECUTING,
+    EXECUTING(false, true),
     /**
      * Only for {@link FlowTaskInstance}, instance is waitting for approval
      */
-    PENDING,
-    WAIT_FOR_CONFIRM,
+    PENDING(false, false),
+
+    WAIT_FOR_CONFIRM(false, true),
     /**
      * Process node is terminated
      */
-    CANCELLED,
+    CANCELLED(true, false),
     /**
      * Process node execution completes
      */
-    COMPLETED,
+    COMPLETED(true, false),
     /**
      * Process node execution expired
      */
-    EXPIRED,
+    EXPIRED(true, false),
     /**
      * Process node execution failed
      */
-    FAILED;
+    FAILED(true, false);
+
+    private final boolean finalStatus;
+    private final boolean executing;
+
+    FlowNodeStatus(boolean finalStatus, boolean executing) {
+        this.finalStatus = finalStatus;
+        this.executing = executing;
+    }
 
     public static Set<FlowNodeStatus> getExecutingStatuses() {
-        return new HashSet<>(Arrays.asList(FlowNodeStatus.EXECUTING, FlowNodeStatus.WAIT_FOR_CONFIRM));
+        return Arrays.stream(FlowNodeStatus.values()).filter(FlowNodeStatus::isExecuting).collect(Collectors.toSet());
     }
 
     public static Set<FlowNodeStatus> getExecutingAndFinalStatuses() {
-        Set<FlowNodeStatus> executingAndFinalStatuses = getExecutingStatuses();
-        executingAndFinalStatuses.addAll(getFinalStatuses());
-        return executingAndFinalStatuses;
+        return Arrays.stream(FlowNodeStatus.values()).filter(t -> t.isExecuting() || t.isFinalStatus()).collect(
+                Collectors.toSet());
     }
 
     public static Set<FlowNodeStatus> getFinalStatuses() {
-        return new HashSet<>(Arrays.asList(FlowNodeStatus.CANCELLED, FlowNodeStatus.COMPLETED, FlowNodeStatus.EXPIRED,
-                FlowNodeStatus.FAILED));
-    }
-
-    public static boolean isFinalStatus(@NonNull FlowNodeStatus status) {
-        return getFinalStatuses().contains(status);
+        return Arrays.stream(FlowNodeStatus.values()).filter(t -> t.isFinalStatus()).collect(Collectors.toSet());
     }
 
-    public boolean isFinalStatus() {
-        return getFinalStatuses().contains(this);
+    public static Set<FlowNodeStatus> getNotFinalStatuses() {
+        return Arrays.stream(FlowNodeStatus.values()).filter(t -> !t.isFinalStatus()).collect(Collectors.toSet());
     }
 
 }

server/odc-service/src/main/java/com/oceanbase/odc/service/schedule/ScheduleService.java

@@ -441,11 +441,10 @@ public String getLog(Long scheduleId, Long taskId, OdcTaskLogLevel logLevel) {
     }
 
     public boolean hasExecutingAsyncTask(ScheduleEntity schedule) {
-        Set<Long> executingTaskIds = serviceTaskRepository.findByScheduleIdAndTaskType(
-                schedule.getId(), TaskType.ASYNC).stream()
-                .filter(entity -> !FlowNodeStatus.isFinalStatus(entity.getStatus()))
-                .map(ServiceTaskInstanceEntity::getTargetTaskId).collect(
-                        Collectors.toSet());
+        Set<Long> executingTaskIds = serviceTaskRepository.findByScheduleIdAndTaskTypeAndStatusIn(schedule.getId(),
+                TaskType.ASYNC, FlowNodeStatus.getNotFinalStatuses()).stream().map(
+                        ServiceTaskInstanceEntity::getTargetTaskId)
+                .collect(Collectors.toSet());
         List<TaskEntity> taskEntities = taskRepository.findByIdIn(executingTaskIds);
         for (TaskEntity taskEntity : taskEntities) {
             Long timeoutMillis = JsonUtils.fromJson(taskEntity.getParametersJson(), DatabaseChangeParameters.class)