Improve tasks executions (spring-cloud#5747)

* Reduce the number of sub queries for tasks/executions

* Added CSVLoader and TaskExecutionQueryIT to test Performance improvement of thinexecutions.

* Added regression check on thin execution performance.

* Remove action that fails in CI-PR

* Improved manifest query.
Updated formatting and copyright dates.

Author: Corneil du Plessis

.github/workflows/ci-it-performance.yml

@@ -0,0 +1,60 @@
+name: CI IT Performance
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: '0 6 * * 1'
+
+jobs:
+  test:
+    name: Performance IT
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+    - uses: actions/setup-java@v3
+      with:
+        java-version: '8'
+        distribution: 'liberica'
+    - uses: jvalkeal/setup-maven@v1
+      with:
+        maven-version: 3.8.8
+        maven-mirror: 'https://dlcdn.apache.org/maven/maven-3/'
+    - name: Login dockerhub
+      uses: docker/login-action@v3
+      with:
+        username: ${{ secrets.DOCKERHUB_USERNAME }}
+        password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+    - name: Run Performance IT
+      run: |
+        mvn \
+          -s .settings.xml \
+          -pl spring-cloud-dataflow-server \
+          -Dgroups=performance \
+          -Pfailsafe \
+          --batch-mode \
+          integration-test
+    - name: Test Report
+      uses: dorny/test-reporter@v1
+      if: ${{ success() || failure() }}
+      with:
+        name: Integration Tests
+        path: '**/failsafe-reports/*IT.xml'
+        reporter: java-junit
+        list-tests: failed
+    - name: 'Action: Upload Unit Test Results'
+      if: ${{ always() }}
+      uses: actions/upload-artifact@v3
+      with:
+        name: ${{ matrix.group }}-test-results-surefire
+        path: './**/target/surefire-reports/**/*.*'
+        retention-days: 7
+        if-no-files-found: ignore
+    - name: 'Action: Upload Integration Test Results'
+      if: ${{ always() }}
+      uses: actions/upload-artifact@v3
+      with:
+        name: ${{ matrix.group }}-test-results-failsafe
+        path: './**/target/failsafe-reports/**/*.*'
+        retention-days: 7
+        if-no-files-found: ignore

.github/workflows/ci-pr.yml

@@ -8,37 +8,22 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
-    # cache maven repo
     - uses: jlumbroso/free-disk-space@main
       with:
         tool-cache: false
         dotnet: false
         docker-images: false
         swap-storage: false
         large-packages: false
-    # jdk8
     - uses: actions/setup-java@v3
       with:
         java-version: '8'
         distribution: 'liberica'
-    - uses: jvalkeal/setup-maven@v1
-      with:
-        maven-version: 3.8.8
-        maven-mirror: 'https://dlcdn.apache.org/maven/maven-3/'
-      # build
     - name: Build
       shell: bash
       timeout-minutes: 75
       run: |
-        mvn -B -s .github/settings.xml clean install
-    - name: Test Report
-      uses: dorny/test-reporter@v1
-      if: ${{ success() || failure() }}
-      with:
-        name: Unit Tests
-        path: '**/surefire-reports/*.xml'
-        reporter: java-junit
-        list-tests: failed
+        ./mvnw -B -s .github/settings.xml clean install
     - name: Capture Test Results
       if: failure()
       uses: actions/upload-artifact@v3

spring-cloud-dataflow-docs/src/main/asciidoc/configuration.adoc

@@ -431,9 +431,10 @@ access the REST API. To do so, retrieve an
 OAuth2 Access Token from your OAuth2 provider and pass that access token to
 the REST Api by using the *Authorization* HTTP header, as follows:
 
-```
+[source, shell]
+----
 $ curl -H "Authorization: Bearer <ACCESS_TOKEN>" http://localhost:9393/ -H 'Accept: application/json'
-```
+----
 
 [[configuration-security-customizing-authorization]]
 ==== Customizing Authorization
@@ -659,6 +660,8 @@ spring:
             - POST   /tasks/executions/*             => hasRole('ROLE_DEPLOY')
             - DELETE /tasks/executions/*             => hasRole('ROLE_DESTROY')
 
+            - GET    /tasks/thinexecutions           => hasRole('ROLE_VIEW')
+
             # Task Schedules
 
             - GET    /tasks/schedules                => hasRole('ROLE_VIEW')

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/batch/JdbcSearchableJobExecutionDao.java

@@ -44,11 +44,13 @@
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowCallbackHandler;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.support.incrementer.AbstractDataFieldMaxValueIncrementer;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Dave Syer
@@ -117,7 +119,9 @@ public class JdbcSearchableJobExecutionDao extends JdbcJobExecutionDao implement
 	private static final String GET_EXECUTION_BY_ID_5 = "SELECT JOB_EXECUTION_ID, START_TIME, END_TIME, STATUS, EXIT_CODE, EXIT_MESSAGE, CREATE_TIME, LAST_UPDATED, VERSION"
 			+ " from %PREFIX%JOB_EXECUTION where JOB_EXECUTION_ID = ?";
 
-	private static final String FROM_CLAUSE_TASK_TASK_BATCH = "%PREFIX%TASK_BATCH B";
+	private static final String FROM_CLAUSE_TASK_TASK_BATCH = "%TASK_PREFIX%TASK_BATCH B";
+
+	private static final String GET_JOB_EXECUTIONS_BY_TASK_IDS = "SELECT JOB_EXECUTION_ID, TASK_EXECUTION_ID from %TASK_PREFIX%TASK_BATCH WHERE TASK_EXECUTION_ID in (?)";
 
 	private PagingQueryProvider allExecutionsPagingQueryProvider;
 
@@ -142,6 +146,7 @@ public class JdbcSearchableJobExecutionDao extends JdbcJobExecutionDao implement
 	private DataSource dataSource;
 
 	private BatchVersion batchVersion;
+	private String taskTablePrefix;
 
 	public JdbcSearchableJobExecutionDao() {
 		this(BatchVersion.BATCH_4);
@@ -161,6 +166,10 @@ public void setDataSource(DataSource dataSource) {
 		this.dataSource = dataSource;
 	}
 
+	public void setTaskTablePrefix(String taskTablePrefix) {
+		this.taskTablePrefix = taskTablePrefix;
+	}
+
 	/**
 	 * @see JdbcJobExecutionDao#afterPropertiesSet()
 	 */
@@ -190,12 +199,15 @@ protected long getNextKey() {
 		byJobInstanceIdWithStepCountPagingQueryProvider = getPagingQueryProvider(FIELDS_WITH_STEP_COUNT, null,
 				JOB_INSTANCE_ID_FILTER);
 		byTaskExecutionIdWithStepCountPagingQueryProvider = getPagingQueryProvider(FIELDS_WITH_STEP_COUNT,
-				FROM_CLAUSE_TASK_TASK_BATCH, TASK_EXECUTION_ID_FILTER);
+				getTaskQuery(FROM_CLAUSE_TASK_TASK_BATCH), TASK_EXECUTION_ID_FILTER);
 
 		super.afterPropertiesSet();
 
 	}
 
+	protected String getTaskQuery(String base) {
+		return StringUtils.replace(base, "%TASK_PREFIX%", taskTablePrefix);
+	}
 	@Override
 	public List<JobExecution> findJobExecutions(JobInstance job) {
 		Assert.notNull(job, "Job cannot be null.");
@@ -517,6 +529,25 @@ public List<JobExecutionWithStepCount> getJobExecutionsWithStepCount(int start,
 		}
 	}
 
+	@Override
+	public Map<Long, Set<Long>> getJobExecutionsByTaskIds(Collection<Long> ids) {
+		JdbcOperations jdbcTemplate = getJdbcTemplate();
+		String strIds = StringUtils.collectionToCommaDelimitedString(ids);
+
+		String sql = getTaskQuery(GET_JOB_EXECUTIONS_BY_TASK_IDS).replace("?", strIds);
+		return jdbcTemplate.query(sql,
+                rs -> {
+            	    final Map<Long, Set<Long>> results = new HashMap<>();
+                	while (rs.next()) {
+                    	Long taskExecutionId = rs.getLong("TASK_EXECUTION_ID");
+                    	Long jobExecutionId = rs.getLong("JOB_EXECUTION_ID");
+                    	Set<Long> jobs = results.computeIfAbsent(taskExecutionId, aLong -> new HashSet<>());
+                    	jobs.add(jobExecutionId);
+                	}
+                	return results;
+				});
+	}
+
 	@Override
 	public void saveJobExecution(JobExecution jobExecution) {
 		throw new UnsupportedOperationException("SearchableJobExecutionDao is read only");

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/batch/JobService.java

@@ -17,6 +17,8 @@
 
 import java.util.Collection;
 import java.util.Date;
+import java.util.Map;
+import java.util.Set;
 
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.Job;
@@ -409,4 +411,11 @@ Collection<JobExecutionWithStepCount> listJobExecutionsForJobWithStepCount(Date
 	 */
 	@Deprecated
 	Collection<JobExecutionWithStepCount> listJobExecutionsForJobWithStepCountFilteredByTaskExecutionId(int taskExecutionId, int start, int count);
+
+	/**
+	 * Returns a collection job execution ids given a collection of task execution ids that is mapped by id.
+	 * @param taskExecutionId
+	 * @return
+	 */
+	Map<Long, Set<Long>> getJobExecutionIdsByTaskExecutionIds(Collection<Long> taskExecutionId);
 }

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/batch/SearchableJobExecutionDao.java

@@ -18,14 +18,16 @@
 import java.util.Collection;
 import java.util.Date;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.repository.dao.JobExecutionDao;
 
 /**
  * @author Dave Syer
- *
+ * @author Corneil du Plessis
  */
 public interface SearchableJobExecutionDao extends JobExecutionDao {
 
@@ -99,6 +101,11 @@ public interface SearchableJobExecutionDao extends JobExecutionDao {
 	 */
 	List<JobExecutionWithStepCount> getJobExecutionsWithStepCount(int start, int count);
 
+	/**
+	 * @param ids the set of task execution ids.
+	 * @return Map with the TaskExecution id as the key and the set of job execution ids as values.
+	 */
+	Map<Long, Set<Long>> getJobExecutionsByTaskIds(Collection<Long> ids);
 	/**
 	 * Gets count of job executions.
 	 *

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/batch/SimpleJobService.java

@@ -25,6 +25,7 @@
 import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
 import java.util.Set;
@@ -468,6 +469,11 @@ public Collection<JobExecutionWithStepCount> listJobExecutionsForJobWithStepCoun
 		return jobExecutionDao.getJobExecutionsWithStepCountFilteredByTaskExecutionId(taskExecutionId, start, count);
 	}
 
+	@Override
+	public Map<Long, Set<Long>> getJobExecutionIdsByTaskExecutionIds(Collection<Long> taskExecutionIds) {
+		return this.jobExecutionDao.getJobExecutionsByTaskIds(taskExecutionIds);
+	}
+
 	private List<JobExecution> getJobExecutions(String jobName, BatchStatus status, int pageOffset, int pageSize) {
 		if (StringUtils.isEmpty(jobName)) {
 			if (status != null) {

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/batch/SimpleJobServiceFactoryBean.java

@@ -73,6 +73,7 @@ public class SimpleJobServiceFactoryBean implements FactoryBean<JobService>, Ini
 	private String databaseType;
 
 	private String tablePrefix = AbstractJdbcBatchMetadataDao.DEFAULT_TABLE_PREFIX;
+	private String taskTablePrefix = "TASK_";
 
 	private DataFieldMaxValueIncrementerFactory incrementerFactory;
 
@@ -165,6 +166,10 @@ public void setTablePrefix(String tablePrefix) {
 		this.tablePrefix = tablePrefix;
 	}
 
+	public void setTaskTablePrefix(String taskTablePrefix) {
+		this.taskTablePrefix = taskTablePrefix;
+	}
+
 	/**
 	 * Sets the {@link JobServiceContainer} for the service.
 	 * @param jobServiceContainer the JobServiceContainer for this service.
@@ -270,6 +275,7 @@ protected SearchableJobExecutionDao createJobExecutionDao() throws Exception {
 		dao.setJobExecutionIncrementer(incrementerFactory.getIncrementer(databaseType, tablePrefix
 				+ "JOB_EXECUTION_SEQ"));
 		dao.setTablePrefix(tablePrefix);
+		dao.setTaskTablePrefix(taskTablePrefix);
 		dao.setClobTypeToUse(determineClobTypeToUse(this.databaseType));
 		dao.setExitMessageLength(maxVarCharLength);
 		dao.afterPropertiesSet();

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/controller/TaskExecutionController.java

@@ -19,10 +19,13 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
@@ -376,18 +379,34 @@ public void stop(
 	}
 
 	private Page<TaskJobExecutionRel> getPageableRelationships(Page<AggregateTaskExecution> taskExecutions, Pageable pageable) {
-		List<TaskJobExecutionRel> taskJobExecutionRels = new ArrayList<>();
-		for (AggregateTaskExecution taskExecution : taskExecutions.getContent()) {
-			TaskManifest taskManifest = this.taskExecutionService.findTaskManifestById(taskExecution.getExecutionId(), taskExecution.getSchemaTarget());
-			taskManifest = this.taskSanitizer.sanitizeTaskManifest(taskManifest);
-			List<Long> jobExecutionIds = new ArrayList<>(
-					this.explorer.getJobExecutionIdsByTaskExecutionId(taskExecution.getExecutionId(), taskExecution.getSchemaTarget()));
-			taskJobExecutionRels
-					.add(new TaskJobExecutionRel(sanitizeTaskExecutionArguments(taskExecution),
-							jobExecutionIds,
-							taskManifest, getCtrTaskJobExecution(taskExecution, jobExecutionIds)));
-		}
-		return new PageImpl<>(taskJobExecutionRels, pageable, taskExecutions.getTotalElements());
+		final Map<String, TaskJobExecutionRel> taskJobExecutionRelMap = new HashMap<>();
+		Map<String, List<AggregateTaskExecution>> schemaGroups = taskExecutions.getContent()
+			.stream()
+			.collect(Collectors.groupingBy(AggregateTaskExecution::getSchemaTarget));
+		schemaGroups.forEach((schemaTarget,aggregateTaskExecutions) -> {
+			Map<Long, AggregateTaskExecution> taskMap = aggregateTaskExecutions.stream().collect(Collectors.toMap(AggregateTaskExecution::getExecutionId, Function.identity()));
+			Set<Long> ids = taskMap.keySet();
+			Map<Long, TaskManifest> manifests = this.taskExecutionService.findTaskManifestByIds(ids, schemaTarget);
+			Map<Long, Set<Long>> jobExecutionIdMap = this.taskJobService.getJobExecutionIdsByTaskExecutionIds(ids, schemaTarget);
+			taskMap.values().forEach(taskExecution -> {
+				long id = taskExecution.getExecutionId();
+				TaskManifest taskManifest = manifests.get(id);
+				if(taskManifest != null) {
+					taskManifest = this.taskSanitizer.sanitizeTaskManifest(taskManifest);
+				}
+				Set<Long> jobIds = jobExecutionIdMap.computeIfAbsent(id, aLong -> new HashSet<>());
+				List<Long> jobExecutionIds = new ArrayList<>(jobIds);
+				TaskJobExecutionRel rel = new TaskJobExecutionRel(sanitizeTaskExecutionArguments(taskExecution),
+					jobExecutionIds,
+					taskManifest, getCtrTaskJobExecution(taskExecution, jobExecutionIds));
+				taskJobExecutionRelMap.put(schemaTarget + ":" + id, rel);
+			});
+		});
+		List<TaskJobExecutionRel> taskJobExecutionContent = taskExecutions.stream()
+			.map(aggregateTaskExecution -> taskJobExecutionRelMap.get(aggregateTaskExecution.getSchemaTarget() + ":" + aggregateTaskExecution.getExecutionId()))
+			.filter(Objects::nonNull)
+			.collect(Collectors.toList());
+		return new PageImpl<>(taskJobExecutionContent, pageable, taskExecutions.getTotalElements());
 	}
 
 
@@ -405,7 +424,7 @@ private TaskJobExecution getCtrTaskJobExecution(AggregateTaskExecution taskExecu
 		TaskDefinition taskDefinition = this.taskDefinitionRepository.findByTaskName(taskExecution.getTaskName());
 		if (taskDefinition != null) {
 			TaskParser parser = new TaskParser(taskExecution.getTaskName(), taskDefinition.getDslText(), true, false);
-			if (jobExecutionIds.size() > 0 && parser.parse().isComposed()) {
+			if (!jobExecutionIds.isEmpty() && parser.parse().isComposed()) {
 				try {
 					taskJobExecution = this.taskJobService.getJobExecution(jobExecutionIds.toArray(new Long[0])[0], taskExecution.getSchemaTarget());
 				} catch (NoSuchJobExecutionException noSuchJobExecutionException) {