Fix a subworkflow performance issue without selecting tasks

core/src/main/java/com/netflix/conductor/core/execution/AsyncSystemTaskExecutor.java

@@ -102,7 +102,9 @@ public void execute(WorkflowSystemTask systemTask, String taskId) {
         // if we are here the Task object is updated and needs to be persisted regardless of an
         // exception
         try {
-            WorkflowModel workflow = executionDAOFacade.getWorkflowModel(workflowId, true);
+            WorkflowModel workflow =
+                    executionDAOFacade.getWorkflowModel(
+                            workflowId, systemTask.isTaskRetrievalRequired());
 
             if (workflow.getStatus().isTerminal()) {
                 LOGGER.info(

core/src/main/java/com/netflix/conductor/core/execution/tasks/SubWorkflow.java

@@ -216,4 +216,14 @@ private void updateTaskStatus(WorkflowModel subworkflow, TaskModel task) {
             }
         }
     }
+
+    /**
+     * We don't need the tasks when retrieving the workflow data.
+     *
+     * @return false
+     */
+    @Override
+    public boolean isTaskRetrievalRequired() {
+        return false;
+    }
 }

core/src/main/java/com/netflix/conductor/core/execution/tasks/WorkflowSystemTask.java

@@ -95,6 +95,17 @@ public String getTaskType() {
         return taskType;
     }
 
+    /**
+     * Default to true for retrieving tasks when retrieving workflow data. Some cases (e.g.
+     * subworkflows) might not need the tasks at all, and by setting this to false in that case, you
+     * can get a solid performance gain.
+     *
+     * @return true for retrieving tasks when getting workflow
+     */
+    public boolean isTaskRetrievalRequired() {
+        return true;
+    }
+
     @Override
     public String toString() {
         return taskType;

core/src/test/groovy/com/netflix/conductor/core/execution/AsyncSystemTaskExecutorTest.groovy

@@ -52,6 +52,8 @@ class AsyncSystemTaskExecutorTest extends Specification {
         workflowExecutor = Mock(WorkflowExecutor.class)
 
         workflowSystemTask = Mock(WorkflowSystemTask.class)
+        then:
+        workflowSystemTask.isTaskRetrievalRequired() >> true
 
         properties.taskExecutionPostponeDuration = Duration.ofSeconds(1)
         properties.systemTaskWorkerCallbackDuration = Duration.ofSeconds(1)
@@ -87,7 +89,7 @@ class AsyncSystemTaskExecutorTest extends Specification {
 
         then:
         1 * executionDAOFacade.getTaskModel(task1Id) >> task1
-        1 * executionDAOFacade.getWorkflowModel(workflowId, true) >> workflow
+        1 * executionDAOFacade.getWorkflowModel(workflowId, subWorkflowTask.isTaskRetrievalRequired()) >> workflow
         1 * workflowExecutor.startWorkflow(*_) >> subWorkflowId
         1 * workflowExecutor.getWorkflow(subWorkflowId, false) >> subWorkflow