From 0e8c26e943e4f3f0d6c18fb81a053b0f21994f08 Mon Sep 17 00:00:00 2001 From: trisberg Date: Thu, 21 Aug 2008 12:30:15 +0000 Subject: [PATCH] BATCH-760: first cut of JdbcPagingItemReader - works for HSQL, need to test other platforms --- .../item/database/JdbcPagingItemReader.java | 327 ++++++++++++++++++ .../JdbcPagingItemReaderCommonTests.java | 63 ++++ .../JdbcPagingItemReaderIntegrationTests.java | 47 +++ ...dbcPagingItemReaderCommonTests-context.xml | 26 ++ 4 files changed, 463 insertions(+) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderIntegrationTests.java create mode 100644 spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests-context.xml diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java new file mode 100644 index 000000000..7c007cb75 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java @@ -0,0 +1,327 @@ +package org.springframework.batch.item.database; + +import org.springframework.batch.item.support.AbstractItemReaderItemStream; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.ClassUtils; +import org.springframework.util.Assert; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.RowCallbackHandler; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.jdbc.core.simple.ParameterizedRowMapper; +import org.springframework.jdbc.support.JdbcUtils; +import org.springframework.jdbc.support.MetaDataAccessException; +import org.springframework.dao.InvalidDataAccessResourceUsageException; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import javax.sql.DataSource; +import java.util.List; +import java.util.ArrayList; +import java.sql.ResultSet; +import java.sql.SQLException; + +/** + * {@link org.springframework.batch.item.ItemReader} for reading database records using JDBC in a paging + * fashion. + * + * It executes the SQL built from values specified for {@link #setSelectClause(String)} (String)}, + * {@link #setFromClause(String)} (String)} and {@link #setWhereClause(String)} (String)} to retrieve requested data. + * The query is executed using paged requests of a size specified in {@link #setPageSize(int)}. Additional pages + * are requested when needed as {@link #read()} method is called, returning an object corresponding to current position. + * + * The performance of the paging depends on the database specific features available to limit the number of returned rows. + * + * Setting a fairly large page size and using a commit interval that matches the page size should provide + * better performance. + * + * The implementation is *not* thread-safe. + * + * @author Thomas Risberg + * @since 2.0 + */ +public class JdbcPagingItemReader extends AbstractItemReaderItemStream implements InitializingBean { + + protected Log logger = LogFactory.getLog(getClass()); + + private DataSource dataSource; + + private SimpleJdbcTemplate simpleJdbcTemplate; + + private ParameterizedRowMapper parameterizedRowMapper; + + private String databaseProductName; + + private String selectClause; + + private String fromClause; + + private String whereClause; + + private String sortKey; + + private String orderClause; + + private String firstPageSql; + + private String remainingPagesSql; + + private boolean initialized = false; + + private int current = 0; + + private int page = 0; + + private int pageSize = 10; + + private Object startAfterValue; + + private List results; + + public JdbcPagingItemReader() { + setName(ClassUtils.getShortName(JdbcPagingItemReader.class)); + } + + public void setDataSource(DataSource dataSource) { + this.dataSource = dataSource; + } + + /** + * @param selectClause SELECT clause part of SQL query string + */ + public void setSelectClause(String selectClause) { + String keyWord = "select "; + String temp = selectClause.trim(); + if (temp.toLowerCase().startsWith(keyWord) && temp.length() > keyWord.length()) { + this.selectClause = temp.substring(keyWord.length()); + } + else { + this.selectClause = temp; + } + } + + /** + * @param fromClause FROM clause part of SQL query string + */ + public void setFromClause(String fromClause) { + String keyWord = "from "; + String temp = fromClause.trim(); + if (temp.toLowerCase().startsWith(keyWord) && temp.length() > keyWord.length()) { + this.fromClause = temp.substring(keyWord.length()); + } + else { + this.fromClause = temp; + } + } + + /** + * @param whereClause WHERE clause part of SQL query string + */ + public void setWhereClause(String whereClause) { + String keyWord = "where "; + String temp = whereClause.trim(); + if (temp.toLowerCase().startsWith(keyWord) && temp.length() > keyWord.length()) { + this.whereClause = temp.substring(keyWord.length()); + } + else { + this.whereClause = temp; + } + } + + /** + * @param sortKey key to use to sort and limit page content + */ + public void setSortKey(String sortKey) { + this.sortKey = sortKey; + } + + /** + * The number of rows to retreive at a time. + * + * @param pageSize the number of rows to fetch per page + */ + public void setPageSize(int pageSize) { + this.pageSize = pageSize; + } + + /** + * The row mapper implementation to be used by this reader + * + * @param parameterizedRowMapper a {@link org.springframework.jdbc.core.simple.ParameterizedRowMapper} implementation + */ + public void setParameterizedRowMapper(ParameterizedRowMapper parameterizedRowMapper) { + this.parameterizedRowMapper = parameterizedRowMapper; + } + + @Override + public void afterPropertiesSet() throws Exception { + Assert.notNull(dataSource); + Assert.hasLength(selectClause, "selectClause must be specified"); + Assert.hasLength(fromClause, "fromClause must be specified"); + Assert.hasLength(sortKey, "sortKey must be specified"); + Assert.isTrue(pageSize > 0, "pageSize must be greater than zero"); + JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource); + jdbcTemplate.setMaxRows(pageSize); + this.simpleJdbcTemplate = new SimpleJdbcTemplate(jdbcTemplate); + initializeSqlStatements(); + } + + private void initializeSqlStatements() throws MetaDataAccessException { + this.databaseProductName = JdbcUtils.commonDatabaseName( + JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductName").toString()); + String topClause = ""; + String limitClause = ""; + if ("DB2".equals(databaseProductName)) { + limitClause = " FETCH FIRST " + pageSize + " ROWS ONLY"; + } + else if ("Oracle".equals(databaseProductName)) { + limitClause = (whereClause == null ? " WHERE " : " AND ") + " ROWNUM <= " + pageSize; + } + else if ("MySQL".equals(databaseProductName) || "PostgreSQL".equals(databaseProductName)) { + limitClause = " LIMIT " + pageSize; + } + else if ("Microsoft SQL Server".equals(databaseProductName) || "Sybase".equals(databaseProductName) || + "HSQL Database Engine".equals(databaseProductName)) { + topClause = "TOP " + pageSize + " "; + } + else if ("Apache Derby".equals(databaseProductName)) { + String version = JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseVersion").toString(); + if ("10.4.1.3".compareTo(version) > 0) { + throw new InvalidDataAccessResourceUsageException(databaseProductName + " version " + version + " is not supported"); + } + // Derby doesn't support TOP or LIMIT -- maxRows will limit the rows retrieved + } + else { + throw new InvalidDataAccessResourceUsageException(databaseProductName + " is not a supported database"); + } + this.orderClause = " ORDER BY SORT_KEY"; + this.firstPageSql = "SELECT " + topClause + selectClause + ", " + sortKey + " AS SORT_KEY" + + " FROM " + fromClause + (whereClause == null ? "" : " WHERE " + whereClause) + orderClause + limitClause; + this.remainingPagesSql = "SELECT " + topClause + selectClause + ", " + sortKey + " AS SORT_KEY" + + " FROM " + fromClause + " WHERE " + sortKey + " > ?" + + (whereClause == null ? "" : " AND " + whereClause) + orderClause + limitClause; + } + + + @Override + @SuppressWarnings("unchecked") + protected T doRead() throws Exception { + + if (results == null || current >= pageSize) { + + if (results == null) { + results = new ArrayList(); + } + else { + results.clear(); + } + + if (page == 0) { + if (logger.isDebugEnabled()) { + logger.debug("SQL used for reading first page: [" + firstPageSql + "]"); + } + simpleJdbcTemplate.getJdbcOperations().query(firstPageSql, + new RowCallbackHandler() { + public void processRow(ResultSet rs) throws SQLException { + startAfterValue = rs.getObject(1); + results.add(parameterizedRowMapper.mapRow(rs, results.size())); + } + }); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("SQL used for reading remaining pages: [" + remainingPagesSql + "]"); + } + simpleJdbcTemplate.getJdbcOperations().query(remainingPagesSql, + new Object[] {startAfterValue}, + new RowCallbackHandler() { + public void processRow(ResultSet rs) throws SQLException { + startAfterValue = rs.getObject(1); + results.add(parameterizedRowMapper.mapRow(rs, results.size())); + } + }); + } + + if (current >= pageSize) { + current = 0; + } + page++; + } + + if (current < results.size()) { + return results.get(current++); + } + else { + return null; + } + + } + + @Override + protected void doOpen() throws Exception { + + Assert.state(!initialized, "Cannot open an already opened ItemReader, call close first"); + + initialized = true; + + } + + @Override + protected void doClose() throws Exception { + + initialized = false; + + } + + + @Override + protected void jumpToItem(int itemIndex) throws Exception { + + page = itemIndex / pageSize; + current = itemIndex % pageSize; + + int lastRow = (page * pageSize) - 1; + + logger.debug("Jumping to page " + page + " and index " + current); + + if (page > 0) { + + String windowClause = ""; + String topClause = ""; + String limitClause = ""; + if ("DB2".equals(databaseProductName) || "Oracle".equals(databaseProductName) || + "Microsoft SQL Server".equals(databaseProductName) || "Sybase".equals(databaseProductName) || + "Apache Derby".equals(databaseProductName)) { + windowClause = "ROW_NUMBER() OVER (ORDER BY " + sortKey + " ASC) AS ROW_NUMBER"; + } + else if ("HSQL Database Engine".equals(databaseProductName)) { + topClause = "LIMIT " + lastRow + " 1 "; + } + else if ("MySQL".equals(databaseProductName) || "PostgreSQL".equals(databaseProductName) || + "HSQL Database Engine".equals(databaseProductName)) { + limitClause = " LIMIT 1 OFFSET " + lastRow; + } + + String jumpToItemSql = + (windowClause.length() > 0 ? "SELECT * FROM ( " : "") + + "SELECT " + (topClause.length() > 0 ? topClause : "") + sortKey + " AS SORT_KEY" + + (windowClause.length() > 0 ? ", " + windowClause : "") + + " FROM " + fromClause + (whereClause == null ? "" : " WHERE " + whereClause) + + (windowClause.length() > 0 ? ") WHERE ROW_NUMBER = " + lastRow : orderClause + limitClause); + + if (logger.isDebugEnabled()) { + logger.debug("SQL used for jumping: [" + jumpToItemSql + "]"); + } + + startAfterValue = simpleJdbcTemplate.getJdbcOperations().queryForObject(jumpToItemSql, + new RowMapper() { + public Object mapRow(ResultSet rs, int i) throws SQLException { + return rs.getObject(1); + } + }); + + } + + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests.java new file mode 100644 index 000000000..bc856bef7 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests.java @@ -0,0 +1,63 @@ +package org.springframework.batch.item.database; + +import org.junit.runner.RunWith; +import org.junit.Test; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.batch.item.CommonItemStreamItemReaderTests; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.sample.Foo; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.simple.ParameterizedRowMapper; + +import javax.persistence.EntityManagerFactory; +import javax.sql.DataSource; +import java.sql.ResultSet; +import java.sql.SQLException; + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class JdbcPagingItemReaderCommonTests extends CommonItemStreamItemReaderTests { + + @Autowired + private DataSource dataSource; + + protected ItemReader getItemReader() throws Exception { + + JdbcPagingItemReader reader = new JdbcPagingItemReader(); + reader.setSelectClause("select ID, NAME, VALUE"); + reader.setFromClause("from T_FOOS"); + reader.setSortKey("ID"); + reader.setDataSource(dataSource); + reader.setParameterizedRowMapper( + new ParameterizedRowMapper() { + public Foo mapRow(ResultSet rs, int i) throws SQLException { + Foo foo = new Foo(); + foo.setId(rs.getInt(1)); + foo.setName(rs.getString(2)); + foo.setValue(rs.getInt(3)); + return foo; + } + } + ); + reader.setPageSize(3); + reader.afterPropertiesSet(); + reader.setSaveState(true); + + return reader; + } + + protected void pointToEmptyInput(ItemReader tested) throws Exception { + JdbcPagingItemReader reader = (JdbcPagingItemReader) tested; + reader.close(new ExecutionContext()); + reader.setSelectClause("select ID, NAME, VALUE"); + reader.setFromClause("from T_FOOS"); + reader.setWhereClause("where id = -1"); + reader.setDataSource(dataSource); + reader.setPageSize(3); + reader.afterPropertiesSet(); + reader.open(new ExecutionContext()); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderIntegrationTests.java new file mode 100644 index 000000000..23f24fd3e --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/JdbcPagingItemReaderIntegrationTests.java @@ -0,0 +1,47 @@ +package org.springframework.batch.item.database; + +import org.junit.runner.RunWith; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.batch.item.sample.Foo; +import org.springframework.batch.item.ItemReader; +import org.springframework.jdbc.core.simple.ParameterizedRowMapper; + +import java.sql.ResultSet; +import java.sql.SQLException; + +/** + * Tests for {@link JpaPagingItemReader}. + * + * @author Thomas Risberg + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(locations = "data-source-context.xml") +public class JdbcPagingItemReaderIntegrationTests extends AbstractDataSourceItemReaderIntegrationTests { + + protected ItemReader createItemReader() throws Exception { + + JdbcPagingItemReader inputSource = new JdbcPagingItemReader(); + inputSource.setSelectClause("select ID, NAME, VALUE"); + inputSource.setFromClause("from T_FOOS"); + inputSource.setSortKey("ID"); + inputSource.setDataSource(dataSource); + inputSource.setParameterizedRowMapper( + new ParameterizedRowMapper() { + public Foo mapRow(ResultSet rs, int i) throws SQLException { + Foo foo = new Foo(); + foo.setId(rs.getInt(1)); + foo.setName(rs.getString(2)); + foo.setValue(rs.getInt(3)); + return foo; + } + } + ); + inputSource.setPageSize(3); + inputSource.afterPropertiesSet(); + inputSource.setSaveState(true); + + return inputSource; + } + +} diff --git a/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests-context.xml b/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests-context.xml new file mode 100644 index 000000000..eee73841b --- /dev/null +++ b/spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/database/JdbcPagingItemReaderCommonTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file