diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilder.java new file mode 100644 index 000000000..4830ccc49 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilder.java @@ -0,0 +1,367 @@ +/* + * Copyright 2017 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. + * 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. + */ +package org.springframework.batch.item.database.builder; + +import java.util.Map; +import javax.sql.DataSource; + +import org.springframework.batch.item.database.JdbcPagingItemReader; +import org.springframework.batch.item.database.Order; +import org.springframework.batch.item.database.PagingQueryProvider; +import org.springframework.batch.item.database.support.AbstractSqlPagingQueryProvider; +import org.springframework.batch.item.database.support.Db2PagingQueryProvider; +import org.springframework.batch.item.database.support.DerbyPagingQueryProvider; +import org.springframework.batch.item.database.support.H2PagingQueryProvider; +import org.springframework.batch.item.database.support.HsqlPagingQueryProvider; +import org.springframework.batch.item.database.support.MySqlPagingQueryProvider; +import org.springframework.batch.item.database.support.OraclePagingQueryProvider; +import org.springframework.batch.item.database.support.PostgresPagingQueryProvider; +import org.springframework.batch.item.database.support.SqlServerPagingQueryProvider; +import org.springframework.batch.item.database.support.SqlitePagingQueryProvider; +import org.springframework.batch.item.database.support.SybasePagingQueryProvider; +import org.springframework.batch.support.DatabaseType; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.support.MetaDataAccessException; +import org.springframework.util.Assert; + +/** + * This is a builder for the {@link JdbcPagingItemReader}. When configuring, either a + * {@link PagingQueryProvider} or the SQL fragments should be provided. If the SQL + * fragments are provided, the metadata from the provided {@link DataSource} will be used + * to create a PagingQueryProvider for you. If both are provided, the PagingQueryProvider + * will be used. + * + * @author Michael Minella + * @since 4.0 + * @see JdbcPagingItemReader + */ +public class JdbcPagingItemReaderBuilder { + + private DataSource dataSource; + + private int fetchSize = JdbcPagingItemReader.VALUE_NOT_SET; + + private PagingQueryProvider queryProvider; + + private RowMapper rowMapper; + + private Map parameterValues; + + private int pageSize = 10; + + private boolean saveState = true; + + private String name; + + private int maxItemCount = Integer.MAX_VALUE; + + private int currentItemCount = 0; + + private String groupClause; + + private String selectClause; + + private String fromClause; + + private String whereClause; + + private Map sortKeys; + + /** + * The {@link DataSource} to query against. Required. + * + * @param dataSource the {@link DataSource} + * @return this instance for method chaining + * @see JdbcPagingItemReader#setDataSource(DataSource) + */ + public JdbcPagingItemReaderBuilder dataSource(DataSource dataSource) { + this.dataSource = dataSource; + + return this; + } + + /** + * A hint to the underlying RDBMS as to how many records to return with each fetch. + * + * @param fetchSize number of records + * @return this instance for method chaining + * @see JdbcPagingItemReader#setFetchSize(int) + */ + public JdbcPagingItemReaderBuilder fetchSize(int fetchSize) { + this.fetchSize = fetchSize; + + return this; + } + + /** + * The {@link RowMapper} used to map the query results to objects. Required. + * + * @param rowMapper a {@link RowMapper} implementation + * @return this instance for method chaining + * @see JdbcPagingItemReader#setRowMapper(RowMapper) + */ + public JdbcPagingItemReaderBuilder rowMapper(RowMapper rowMapper) { + this.rowMapper = rowMapper; + + return this; + } + + /** + * A {@link Map} of values to set on the SQL's prepared statement. + * + * @param parameterValues Map of values + * @return this instance for method chaining + * @see JdbcPagingItemReader#setParameterValues(Map) + */ + public JdbcPagingItemReaderBuilder parameterValues(Map parameterValues) { + this.parameterValues = parameterValues; + + return this; + } + + /** + * The number of records to request per page/query. Defaults to 10. Must be greater + * than zero. + * + * @param pageSize number of items + * @return this instance for method chaining + * @see JdbcPagingItemReader#setPageSize(int) + */ + public JdbcPagingItemReaderBuilder pageSize(int pageSize) { + this.pageSize = pageSize; + + return this; + } + + /** + * Set to false in a multithreaded environment (restart is disabled). If set to true, + * a name is required. Defaults to true. + * + * @param saveState determine if the reader's state should be persisted + * @return this instance for method chaining + * @see JdbcPagingItemReader#setSaveState(boolean) + */ + public JdbcPagingItemReaderBuilder saveState(boolean saveState) { + this.saveState = saveState; + + return this; + } + + /** + * A name used to prevent key collissions while saving the state in the + * {@link org.springframework.batch.item.ExecutionContext} + * + * @param name unique name for this reader instance + * @return this instance for method chaining + * @see JdbcPagingItemReader#setName(String) + */ + public JdbcPagingItemReaderBuilder name(String name) { + this.name = name; + + return this; + } + + /** + * Maximum number of items to read. + * + * @param count number of items + * @return this instance for method chaining + * @see JdbcPagingItemReader#setMaxItemCount(int) + */ + public JdbcPagingItemReaderBuilder maxItemCount(int count) { + this.maxItemCount = count; + + return this; + } + + /** + * The current index of the item to read. + * + * @param count current index + * @return this instance for method chaining + * @see JdbcPagingItemReader#setCurrentItemCount(int) + */ + public JdbcPagingItemReaderBuilder currentItemCount(int count) { + this.currentItemCount = count; + + return this; + } + + /** + * The SQL GROUP BY clause for a db specific @{@link PagingQueryProvider}. + * This is only used if a PaginingQueryProvider is not provided. + * + * @param groupClause the SQL clause + * @return this instance for method chaining + * @see AbstractSqlPagingQueryProvider#setGroupClause(String) + */ + public JdbcPagingItemReaderBuilder groupClause(String groupClause) { + this.groupClause = groupClause; + + return this; + } + + /** + * The SQL SELECT clause for a db specific {@link PagingQueryProvider}. + * This is only used if a PagingQueyrProvider is not provided. + * + * @param selectClause the SQL clause + * @return this instance for method chaining + * @see AbstractSqlPagingQueryProvider#setSelectClause(String) + */ + public JdbcPagingItemReaderBuilder selectClause(String selectClause) { + this.selectClause = selectClause; + + return this; + } + + /** + * The SQL FROM clause for a db specific {@link PagingQueryProvider}. + * This is only used if a PagingQueryProvider is not provided. + * + * @param fromClause the SQL clause + * @return this instance for method chaining + * @see AbstractSqlPagingQueryProvider#setFromClause(String) + */ + public JdbcPagingItemReaderBuilder fromClause(String fromClause) { + this.fromClause = fromClause; + + return this; + } + + /** + * The SQL WHERE clause for a db specific {@link PagingQueryProvider}. + * This is only used if a PagingQueryProvider is not provided. + * + * @param whereClause the SQL clause + * @return this instance for method chaining + * @see AbstractSqlPagingQueryProvider#setWhereClause(String) + */ + public JdbcPagingItemReaderBuilder whereClause(String whereClause) { + this.whereClause = whereClause; + + return this; + } + + /** + * The keys to sort by. These keys must create a unique key. + * + * @param sortKeys keys to sort by and the direction for each. + * @return this instance for method chaining + * @see AbstractSqlPagingQueryProvider#setSortKeys(Map) + */ + public JdbcPagingItemReaderBuilder sortKeys(Map sortKeys) { + this.sortKeys = sortKeys; + + return this; + } + + /** + * A {@link PagingQueryProvider} to provide the queries required. If provided, the + * SQL fragments configured via {@link #selectClause(String)}, + * {@link #fromClause(String)}, {@link #whereClause(String)}, {@link #groupClause}, + * and {@link #sortKeys(Map)} are ignored. + * + * @param provider the db specific query provider + * @return this instance for method chaining + * @see JdbcPagingItemReader#setQueryProvider(PagingQueryProvider) + */ + public JdbcPagingItemReaderBuilder queryProvider(PagingQueryProvider provider) { + this.queryProvider = provider; + + return this; + } + + /** + * Provides a completely built instance of the {@link JdbcPagingItemReader} + * + * @return a {@link JdbcPagingItemReader} + */ + public JdbcPagingItemReader build() { + Assert.isTrue(this.pageSize > 0, "pageSize must be greater than zero"); + Assert.notNull(this.dataSource, "dataSource is required"); + + if(this.saveState) { + Assert.hasText(this.name, + "A name is required when saveState is set to true"); + } + + JdbcPagingItemReader reader = new JdbcPagingItemReader<>(); + + reader.setMaxItemCount(this.maxItemCount); + reader.setCurrentItemCount(this.currentItemCount); + reader.setName(this.name); + reader.setSaveState(this.saveState); + reader.setDataSource(this.dataSource); + reader.setFetchSize(this.fetchSize); + reader.setParameterValues(this.parameterValues); + + if(this.queryProvider == null) { + Assert.hasLength(this.selectClause, "selectClause is required when not providing a PagingQueryProvider"); + Assert.hasLength(this.fromClause, "fromClause is required when not providing a PagingQueryProvider"); + Assert.notEmpty(this.sortKeys, "sortKeys are required when not providing a PagingQueryProvider"); + + reader.setQueryProvider(determineQueryProvider(this.dataSource)); + } + else { + reader.setQueryProvider(this.queryProvider); + } + + reader.setRowMapper(this.rowMapper); + reader.setPageSize(this.pageSize); + + return reader; + } + + private PagingQueryProvider determineQueryProvider(DataSource dataSource) { + + try { + DatabaseType databaseType = DatabaseType.fromMetaData(this.dataSource); + + AbstractSqlPagingQueryProvider provider; + + switch (databaseType) { + + case DERBY: provider = new DerbyPagingQueryProvider(); break; + case DB2: + case DB2VSE: + case DB2ZOS: + case DB2AS400: provider = new Db2PagingQueryProvider(); break; + case H2: provider = new H2PagingQueryProvider(); break; + case HSQL: provider = new HsqlPagingQueryProvider(); break; + case SQLSERVER: provider = new SqlServerPagingQueryProvider(); break; + case MYSQL: provider = new MySqlPagingQueryProvider(); break; + case ORACLE: provider = new OraclePagingQueryProvider(); break; + case POSTGRES: provider = new PostgresPagingQueryProvider(); break; + case SYBASE: provider = new SybasePagingQueryProvider(); break; + case SQLITE: provider = new SqlitePagingQueryProvider(); break; + default: + throw new IllegalArgumentException("Unalbe to determine PaaginQueryProvider type"); + } + + provider.setSelectClause(this.selectClause); + provider.setFromClause(this.fromClause); + provider.setWhereClause(this.whereClause); + provider.setGroupClause(this.groupClause); + provider.setSortKeys(this.sortKeys); + + return provider; + } + catch (MetaDataAccessException e) { + throw new IllegalArgumentException("Unable to determine PagingQueryProvider type", e); + } + } +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilderTests.java new file mode 100644 index 000000000..944dbb1a0 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilderTests.java @@ -0,0 +1,418 @@ +/* + * Copyright 2017 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. + * 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. + */ +package org.springframework.batch.item.database.builder; + +import java.util.HashMap; +import java.util.Map; +import javax.sql.DataSource; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.database.JdbcPagingItemReader; +import org.springframework.batch.item.database.Order; +import org.springframework.batch.item.database.support.AbstractSqlPagingQueryProvider; +import org.springframework.batch.item.database.support.HsqlPagingQueryProvider; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.ByteArrayResource; +import org.springframework.core.io.Resource; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseFactory; +import org.springframework.jdbc.datasource.init.DataSourceInitializer; +import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; +import org.springframework.test.util.ReflectionTestUtils; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +/** + * @author Michael Minella + */ +public class JdbcPagingItemReaderBuilderTests { + + private DataSource dataSource; + + private ConfigurableApplicationContext context; + + @Before + public void setUp() { + this.context = new AnnotationConfigApplicationContext(TestDataSourceConfiguration.class); + this.dataSource = (DataSource) context.getBean("dataSource"); + } + + @After + public void tearDown() { + if(this.context != null) { + this.context.close(); + } + } + + @Test + public void testBasicConfigurationQueryProvider() throws Exception { + Map sortKeys = new HashMap<>(1); + sortKeys.put("ID", Order.DESCENDING); + + AbstractSqlPagingQueryProvider provider = new HsqlPagingQueryProvider(); + provider.setSelectClause("SELECT ID, FIRST, SECOND, THIRD"); + provider.setFromClause("FOO"); + provider.setSortKeys(sortKeys); + + JdbcPagingItemReader reader = new JdbcPagingItemReaderBuilder() + .name("fooReader") + .currentItemCount(1) + .dataSource(this.dataSource) + .queryProvider(provider) + .fetchSize(2) + .maxItemCount(2) + .rowMapper((rs, rowNum) -> new Foo(rs.getInt(1), + rs.getInt(2), + rs.getString(3), + rs.getString(4))) + .build(); + + reader.afterPropertiesSet(); + + ExecutionContext executionContext = new ExecutionContext(); + reader.open(executionContext); + Foo item1 = reader.read(); + assertNull(reader.read()); + reader.update(executionContext); + reader.close(); + + assertEquals(3, item1.getId()); + assertEquals(10, item1.getFirst()); + assertEquals("11", item1.getSecond()); + assertEquals("12", item1.getThird()); + assertTrue((int) ReflectionTestUtils.getField(reader, "fetchSize") == 2); + + assertEquals(2, executionContext.size()); + } + + @Test + public void testBasicConfiguration() throws Exception { + Map sortKeys = new HashMap<>(1); + sortKeys.put("ID", Order.DESCENDING); + + JdbcPagingItemReader reader = new JdbcPagingItemReaderBuilder() + .name("fooReader") + .currentItemCount(1) + .dataSource(this.dataSource) + .maxItemCount(2) + .selectClause("SELECT ID, FIRST, SECOND, THIRD") + .fromClause("FOO") + .sortKeys(sortKeys) + .rowMapper((rs, rowNum) -> new Foo(rs.getInt(1), + rs.getInt(2), + rs.getString(3), + rs.getString(4))) + .build(); + + reader.afterPropertiesSet(); + + reader.open(new ExecutionContext()); + Foo item1 = reader.read(); + assertNull(reader.read()); + + assertEquals(3, item1.getId()); + assertEquals(10, item1.getFirst()); + assertEquals("11", item1.getSecond()); + assertEquals("12", item1.getThird()); + } + + @Test + public void testPageSize() throws Exception { + Map sortKeys = new HashMap<>(1); + sortKeys.put("ID", Order.DESCENDING); + + JdbcPagingItemReader reader = new JdbcPagingItemReaderBuilder() + .name("fooReader") + .dataSource(this.dataSource) + .pageSize(1) + .maxItemCount(2) + .selectClause("SELECT ID, FIRST, SECOND, THIRD") + .fromClause("FOO") + .sortKeys(sortKeys) + .rowMapper((rs, rowNum) -> new Foo(rs.getInt(1), + rs.getInt(2), + rs.getString(3), + rs.getString(4))) + .build(); + + reader.afterPropertiesSet(); + + reader.open(new ExecutionContext()); + Foo item1 = reader.read(); + Foo item2 = reader.read(); + assertNull(reader.read()); + + assertEquals(4, item1.getId()); + assertEquals(13, item1.getFirst()); + assertEquals("14", item1.getSecond()); + assertEquals("15", item1.getThird()); + + assertEquals(3, item2.getId()); + assertEquals(10, item2.getFirst()); + assertEquals("11", item2.getSecond()); + assertEquals("12", item2.getThird()); + } + + @Test + public void testSaveState() throws Exception { + Map sortKeys = new HashMap<>(1); + sortKeys.put("ID", Order.DESCENDING); + + JdbcPagingItemReader reader = new JdbcPagingItemReaderBuilder() + .dataSource(this.dataSource) + .pageSize(1) + .maxItemCount(2) + .selectClause("SELECT ID, FIRST, SECOND, THIRD") + .fromClause("FOO") + .sortKeys(sortKeys) + .saveState(false) + .rowMapper((rs, rowNum) -> new Foo(rs.getInt(1), + rs.getInt(2), + rs.getString(3), + rs.getString(4))) + .build(); + + reader.afterPropertiesSet(); + + ExecutionContext executionContext = new ExecutionContext(); + reader.open(executionContext); + Foo item1 = reader.read(); + Foo item2 = reader.read(); + assertNull(reader.read()); + reader.update(executionContext); + reader.close(); + + assertEquals(4, item1.getId()); + assertEquals(13, item1.getFirst()); + assertEquals("14", item1.getSecond()); + assertEquals("15", item1.getThird()); + + assertEquals(3, item2.getId()); + assertEquals(10, item2.getFirst()); + assertEquals("11", item2.getSecond()); + assertEquals("12", item2.getThird()); + + assertEquals(0, executionContext.size()); + } + + @Test + public void testParameters() throws Exception { + Map sortKeys = new HashMap<>(1); + sortKeys.put("ID", Order.DESCENDING); + + Map parameterValues = new HashMap<>(); + parameterValues.put("min", 1); + parameterValues.put("max", 10); + + JdbcPagingItemReader reader = new JdbcPagingItemReaderBuilder() + .name("fooReader") + .dataSource(this.dataSource) + .pageSize(1) + .maxItemCount(1) + .selectClause("SELECT ID, FIRST, SECOND, THIRD") + .fromClause("FOO") + .whereClause("FIRST > :min AND FIRST < :max") + .sortKeys(sortKeys) + .parameterValues(parameterValues) + .rowMapper((rs, rowNum) -> new Foo(rs.getInt(1), + rs.getInt(2), + rs.getString(3), + rs.getString(4))) + .build(); + + reader.afterPropertiesSet(); + + reader.open(new ExecutionContext()); + Foo item1 = reader.read(); + assertNull(reader.read()); + + assertEquals(2, item1.getId()); + assertEquals(7, item1.getFirst()); + assertEquals("8", item1.getSecond()); + assertEquals("9", item1.getThird()); + } + + @Test + public void testValidation() { + + try { + new JdbcPagingItemReaderBuilder().build(); + fail(); + } + catch (IllegalArgumentException iae) { + assertEquals("dataSource is required", iae.getMessage()); + } + + try { + new JdbcPagingItemReaderBuilder() + .pageSize(-2) + .build(); + fail(); + } + catch (IllegalArgumentException iae) { + assertEquals("pageSize must be greater than zero", iae.getMessage()); + } + + try { + new JdbcPagingItemReaderBuilder() + .pageSize(2) + .build(); + fail(); + } + catch (IllegalArgumentException ise) { + assertEquals("dataSource is required", ise.getMessage()); + } + + try { + new JdbcPagingItemReaderBuilder() + .pageSize(2) + .dataSource(this.dataSource) + .build(); + fail(); + } + catch (IllegalArgumentException ise) { + assertEquals("A name is required when saveState is set to true", ise.getMessage()); + } + + try { + new JdbcPagingItemReaderBuilder() + .saveState(false) + .pageSize(2) + .dataSource(this.dataSource) + .build(); + fail(); + } + catch (IllegalArgumentException ise) { + assertEquals("selectClause is required when not providing a PagingQueryProvider", ise.getMessage()); + } + + try { + new JdbcPagingItemReaderBuilder() + .name("fooReader") + .pageSize(2) + .dataSource(this.dataSource) + .selectClause("SELECT *") + .build(); + fail(); + } + catch (IllegalArgumentException ise) { + assertEquals("fromClause is required when not providing a PagingQueryProvider", ise.getMessage()); + } + + try { + new JdbcPagingItemReaderBuilder() + .saveState(false) + .pageSize(2) + .dataSource(this.dataSource) + .selectClause("SELECT *") + .fromClause("FOO") + .build(); + fail(); + } + catch (IllegalArgumentException ise) { + assertEquals("sortKeys are required when not providing a PagingQueryProvider", ise.getMessage()); + } + } + + public static class Foo { + private int id; + private int first; + private String second; + private String third; + + public Foo(int id, int first, String second, String third) { + this.id = id; + this.first = first; + this.second = second; + this.third = third; + } + + public int getId() { + return id; + } + + public void setId(int id) { + this.id = id; + } + + public int getFirst() { + return first; + } + + public void setFirst(int first) { + this.first = first; + } + + public String getSecond() { + return second; + } + + public void setSecond(String second) { + this.second = second; + } + + public String getThird() { + return third; + } + + public void setThird(String third) { + this.third = third; + } + } + + @Configuration + public static class TestDataSourceConfiguration { + + private static final String CREATE_SQL = "CREATE TABLE FOO (\n" + + "\tID BIGINT IDENTITY NOT NULL PRIMARY KEY ,\n" + + "\tFIRST BIGINT ,\n" + + "\tSECOND VARCHAR(5) NOT NULL,\n" + + "\tTHIRD VARCHAR(5) NOT NULL) ;"; + + private static final String INSERT_SQL = + "INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (1, '2', '3');" + + "INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (4, '5', '6');" + + "INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (7, '8', '9');" + + "INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (10, '11', '12');" + + "INSERT INTO FOO (FIRST, SECOND, THIRD) VALUES (13, '14', '15');"; + + @Bean + public DataSource dataSource() { + return new EmbeddedDatabaseFactory().getDatabase(); + } + + @Bean + public DataSourceInitializer initializer(DataSource dataSource) { + DataSourceInitializer dataSourceInitializer = new DataSourceInitializer(); + dataSourceInitializer.setDataSource(dataSource); + + Resource create = new ByteArrayResource(CREATE_SQL.getBytes()); + Resource insert = new ByteArrayResource(INSERT_SQL.getBytes()); + dataSourceInitializer.setDatabasePopulator(new ResourceDatabasePopulator(create, insert)); + + return dataSourceInitializer; + } + } +}