Use ROW_NUMBER in JdbcAggregateJobQueryDao to improve performance (sp…

…ring-cloud#5575)

* Use ROW_NUMBER in JdbcAggregateJobQueryDao to improve performance

See spring-cloud#5524

spring-cloud-dataflow-core/src/main/java/org/springframework/cloud/dataflow/core/database/support/DatabaseType.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.dataflow.core.database.support;
 
+import java.sql.DatabaseMetaData;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -100,6 +101,24 @@ public static DatabaseType fromProductName(String productName) {
 		}
 	}
 
+	/**
+	 * Determines if the Database that the datasource refers to supports the {@code ROW_NUMBER()} SQL function.
+	 * @param dataSource the datasource pointing to the DB in question
+	 * @return whether the database supports the SQL {@code ROW_NUMBER()} function
+	 * @throws MetaDataAccessException if error occurs
+	 */
+	public static boolean supportsRowNumberFunction(DataSource dataSource) throws MetaDataAccessException {
+		DatabaseType databaseType = DatabaseType.fromMetaData(dataSource);
+		if (databaseType == DatabaseType.H2 || databaseType == DatabaseType.HSQL) {
+			return false;
+		}
+		if (databaseType != DatabaseType.MYSQL) {
+			return true;
+		}
+		int majorVersion = JdbcUtils.extractDatabaseMetaData(dataSource, DatabaseMetaData::getDatabaseMajorVersion);
+		return (majorVersion >= 8);
+	}
+
 	private String getProductName() {
 		return productName;
 	}

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

@@ -45,6 +45,8 @@
 import org.springframework.cloud.dataflow.server.repository.AggregateJobQueryDao;
 import org.springframework.cloud.dataflow.server.repository.JdbcAggregateJobQueryDao;
 import org.springframework.cloud.dataflow.server.service.JobServiceContainer;
+import org.springframework.context.EnvironmentAware;
+import org.springframework.core.env.Environment;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.support.lob.DefaultLobHandler;
@@ -60,7 +62,7 @@
  * @author Dave Syer
  *
  */
-public class SimpleJobServiceFactoryBean implements FactoryBean<JobService>, InitializingBean {
+public class SimpleJobServiceFactoryBean implements FactoryBean<JobService>, InitializingBean, EnvironmentAware {
 
 	private static final Logger logger = LoggerFactory.getLogger(SimpleJobServiceFactoryBean.class);
 
@@ -94,6 +96,8 @@ public class SimpleJobServiceFactoryBean implements FactoryBean<JobService>, Ini
 
 	private SchemaVersionTarget schemaVersionTarget;
 
+	private Environment environment;
+
 	public void setTransactionManager(PlatformTransactionManager transactionManager) {
 		this.transactionManager = transactionManager;
 	}
@@ -132,6 +136,11 @@ public void setMaxVarCharLength(int maxVarCharLength) {
 		this.maxVarCharLength = maxVarCharLength;
 	}
 
+	@Override
+	public void setEnvironment(Environment environment) {
+		this.environment = environment;
+	}
+
 	/**
 	 * Public setter for the {@link DataSource}.
 	 * @param dataSource a {@link DataSource}
@@ -301,10 +310,11 @@ private int determineClobTypeToUse(String databaseType) {
 			return Types.CLOB;
 		}
 	}
+
 	protected AggregateJobQueryDao createAggregateJobQueryDao() throws Exception {
-		AggregateJobQueryDao dao = new JdbcAggregateJobQueryDao(this.dataSource, this.schemaService, this.jobServiceContainer);
-		return dao;
+		return new JdbcAggregateJobQueryDao(this.dataSource, this.schemaService, this.jobServiceContainer, this.environment);
 	}
+
 	/**
 	 * Create a {@link SimpleJobService} from the configuration provided.
 	 *

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/config/AggregateDataFlowTaskConfiguration.java

@@ -55,6 +55,7 @@
 import org.springframework.cloud.task.repository.support.DatabaseType;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
 import org.springframework.jdbc.support.MetaDataAccessException;
 import org.springframework.transaction.PlatformTransactionManager;
 
@@ -129,8 +130,11 @@ public JobExplorerContainer jobExplorerContainer(DataSource dataSource, SchemaSe
 	}
 
 	@Bean
-	public JobServiceContainer jobServiceContainer(DataSource dataSource, PlatformTransactionManager platformTransactionManager, SchemaService schemaService, JobRepositoryContainer jobRepositoryContainer, JobExplorerContainer jobExplorerContainer) {
-		return new JobServiceContainer(dataSource, platformTransactionManager, schemaService, jobRepositoryContainer, jobExplorerContainer);
+	public JobServiceContainer jobServiceContainer(DataSource dataSource, PlatformTransactionManager platformTransactionManager,
+			SchemaService schemaService, JobRepositoryContainer jobRepositoryContainer,
+			JobExplorerContainer jobExplorerContainer, Environment environment) {
+		return new JobServiceContainer(dataSource, platformTransactionManager, schemaService, jobRepositoryContainer,
+				jobExplorerContainer, environment);
 	}
 
 	@Bean
@@ -151,17 +155,13 @@ public TaskDeploymentReader taskDeploymentReader(TaskDeploymentRepository reposi
 	}
 
 	@Bean
-	public AggregateJobQueryDao aggregateJobQueryDao(DataSource dataSource, SchemaService schemaService, JobServiceContainer jobServiceContainer) throws Exception {
-		return new JdbcAggregateJobQueryDao(dataSource, schemaService, jobServiceContainer);
+	public AggregateJobQueryDao aggregateJobQueryDao(DataSource dataSource, SchemaService schemaService,
+			JobServiceContainer jobServiceContainer, Environment environment) throws Exception {
+		return new JdbcAggregateJobQueryDao(dataSource, schemaService, jobServiceContainer, environment);
 	}
 
 	@Bean
 	public TaskBatchDaoContainer taskBatchDaoContainer(DataSource dataSource, SchemaService schemaService) {
 		return new TaskBatchDaoContainer(dataSource, schemaService);
 	}
-
-	@PostConstruct
-	public void setup() {
-		logger.info("created: org.springframework.cloud.dataflow.server.config.AggregateDataFlowContainerConfiguration");
-	}
 }

spring-cloud-dataflow-server-core/src/main/java/org/springframework/cloud/dataflow/server/repository/AggregateJobQueryDao.java

@@ -33,6 +33,7 @@
  * Provides for reading job execution data for Batch 4 and 5 schema versions.
  *
  * @author Corneil du Plessis
+ * @since 2.11.0
  */
 public interface AggregateJobQueryDao {
 	Page<JobInstanceExecutions> listJobInstances(String jobName, Pageable pageable) throws NoSuchJobException;