Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Creates a new builder for the JdbcPagingItemReader

Browse Source 
This commit creates a builder for the JdbcPagingItemReader

Resolves BATCH-2582
This commit is contained in:
Michael Minella
2017-03-10 17:40:25 -06:00
parent a6c4dacbbe
commit 7f94ca2e84
2 changed files with 785 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

367
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilder.java Normal file
View File

@@ -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<T> {
private DataSource dataSource;
private int fetchSize = JdbcPagingItemReader.VALUE_NOT_SET;
private PagingQueryProvider queryProvider;
private RowMapper<T> rowMapper;
private Map<String, Object> 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<String, Order> 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<T> 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<T> 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<T> rowMapper(RowMapper<T> 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<T> parameterValues(Map<String, Object> 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<T> 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<T> 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<T> 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<T> 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<T> currentItemCount(int count) {
this.currentItemCount = count;
return this;
}
/**
* The SQL <code>GROUP BY</code> 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<T> groupClause(String groupClause) {
this.groupClause = groupClause;
return this;
}
/**
* The SQL <code>SELECT</code> 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<T> selectClause(String selectClause) {
this.selectClause = selectClause;
return this;
}
/**
* The SQL <code>FROM</code> 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<T> fromClause(String fromClause) {
this.fromClause = fromClause;
return this;
}
/**
* The SQL <code>WHERE</code> 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<T> whereClause(String whereClause) {
this.whereClause = whereClause;
return this;
}
/**
* The keys to sort by. These keys <em>must</em> 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<T> sortKeys(Map<String, Order> 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<T> queryProvider(PagingQueryProvider provider) {
this.queryProvider = provider;
return this;
}
/**
* Provides a completely built instance of the {@link JdbcPagingItemReader}
*
* @return a {@link JdbcPagingItemReader}
*/
public JdbcPagingItemReader<T> 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<T> 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);
}
}
}

418
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcPagingItemReaderBuilderTests.java Normal file
View File

@@ -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<String, Order> 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<Foo> reader = new JdbcPagingItemReaderBuilder<Foo>()
.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<String, Order> sortKeys = new HashMap<>(1);
sortKeys.put("ID", Order.DESCENDING);
JdbcPagingItemReader<Foo> reader = new JdbcPagingItemReaderBuilder<Foo>()
.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<String, Order> sortKeys = new HashMap<>(1);
sortKeys.put("ID", Order.DESCENDING);
JdbcPagingItemReader<Foo> reader = new JdbcPagingItemReaderBuilder<Foo>()
.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<String, Order> sortKeys = new HashMap<>(1);
sortKeys.put("ID", Order.DESCENDING);
JdbcPagingItemReader<Foo> reader = new JdbcPagingItemReaderBuilder<Foo>()
.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<String, Order> sortKeys = new HashMap<>(1);
sortKeys.put("ID", Order.DESCENDING);
Map<String, Object> parameterValues = new HashMap<>();
parameterValues.put("min", 1);
parameterValues.put("max", 10);
JdbcPagingItemReader<Foo> reader = new JdbcPagingItemReaderBuilder<Foo>()
.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<Foo>().build();
fail();
}
catch (IllegalArgumentException iae) {
assertEquals("dataSource is required", iae.getMessage());
}
try {
new JdbcPagingItemReaderBuilder<Foo>()
.pageSize(-2)
.build();
fail();
}
catch (IllegalArgumentException iae) {
assertEquals("pageSize must be greater than zero", iae.getMessage());
}
try {
new JdbcPagingItemReaderBuilder<Foo>()
.pageSize(2)
.build();
fail();
}
catch (IllegalArgumentException ise) {
assertEquals("dataSource is required", ise.getMessage());
}
try {
new JdbcPagingItemReaderBuilder<Foo>()
.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<Foo>()
.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<Foo>()
.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<Foo>()
.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;
}
}
}