BATCH-760: added remaining database specific implementations and a delegating implementation that uses database metadata to determine which implementation to delegate to

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcPagingItemReader.java

@@ -123,6 +123,7 @@ public class JdbcPagingItemReader<T> extends AbstractItemReaderItemStream<T> imp
 		jdbcTemplate.setMaxRows(pageSize);
 		this.simpleJdbcTemplate = new SimpleJdbcTemplate(jdbcTemplate);
 		Assert.notNull(queryProvider);
+		queryProvider.init(dataSource);
 		this.firstPageSql = queryProvider.generateFirstPageQuery(pageSize);
 		this.remainingPagesSql = queryProvider.generateRemainingPagesQuery(pageSize);
 	}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/Db2PagingQueryProvider.java

@@ -0,0 +1,35 @@
+package org.springframework.batch.item.database.support;
+
+/**
+ * Oracle implementation of a  {@link PagingQueryProvider} using
+ * database specific features.
+ *
+ * @author Thomas Risberg
+ * @since 2.0
+ */
+public class Db2PagingQueryProvider extends SqlWindowingPagingQueryProvider {
+
+	@Override
+	public String generateFirstPageQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(getWhereClause() == null ? "" : " WHERE " + getWhereClause());
+		sql.append(" FETCH FIRST ").append(pageSize).append(" ROWS ONLY");
+
+		return sql.toString();
+	}
+
+	@Override
+	public String generateRemainingPagesQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(" WHERE ").append(getSortKey()).append(" > ?");
+		sql.append(getWhereClause() == null ? "" : " AND " + getWhereClause());
+		sql.append(" FETCH FIRST ").append(pageSize).append(" ROWS ONLY");
+
+		return sql.toString();
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/OraclePagingQueryProvider.java

@@ -0,0 +1,39 @@
+package org.springframework.batch.item.database.support;
+
+import org.springframework.jdbc.support.JdbcUtils;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+
+import javax.sql.DataSource;
+
+/**
+ * Oracle implementation of a  {@link org.springframework.batch.item.database.support.PagingQueryProvider} using
+ * database specific features.
+ *
+ * @author Thomas Risberg
+ * @since 2.0
+ */
+public class OraclePagingQueryProvider extends SqlWindowingPagingQueryProvider {
+
+	@Override
+	public String generateFirstPageQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(" WHERE ROWNUM <= ").append(pageSize);
+		sql.append(getWhereClause() == null ? "" : " AND " + getWhereClause());
+
+		return sql.toString();
+	}
+
+	@Override
+	public String generateRemainingPagesQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(" WHERE ").append(getSortKey()).append(" > ?");
+		sql.append(" AND ROWNUM <= ").append(pageSize);
+		sql.append(getWhereClause() == null ? "" : " AND " + getWhereClause());
+
+		return sql.toString();
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/PostgresPagingQueryProvider.java

@@ -0,0 +1,47 @@
+package org.springframework.batch.item.database.support;
+
+/**
+ * MySQL implementation of a  {@link org.springframework.batch.item.database.support.PagingQueryProvider} using database specific features.
+ *
+ * @author Thomas Risberg
+ * @since 2.0
+ */
+public class PostgresPagingQueryProvider extends AbstractSqlPagingQueryProvider {
+
+	@Override
+	public String generateFirstPageQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(getWhereClause() == null ? "" : " WHERE " + getWhereClause());
+		sql.append(" LIMIT ").append(pageSize);
+
+		return sql.toString();
+	}
+
+	@Override
+	public String generateRemainingPagesQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(" WHERE ").append(getSortKey()).append(" > ?");
+		sql.append(getWhereClause() == null ? "" : " AND " + getWhereClause());
+		sql.append(" LIMIT ").append(pageSize);
+
+		return sql.toString();
+	}
+
+	@Override
+	public String generateJumpToItemQuery(int itemIndex, int pageSize) {
+		int page = itemIndex / pageSize;
+		int offset = (page * pageSize) - 1;
+
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append(getSortKey()).append(" AS SORT_KEY");
+		sql.append(" FROM ").append(getFromClause()).append(getWhereClause() == null ? "" : " WHERE " + getWhereClause());
+		sql.append(" LIMIT ").append(offset).append(" 1");
+
+		return sql.toString();
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProvider.java

@@ -0,0 +1,91 @@
+package org.springframework.batch.item.database.support;
+
+import org.springframework.jdbc.support.JdbcUtils;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+
+import javax.sql.DataSource;
+import java.util.List;
+import java.util.Arrays;
+
+/**
+ * Generic Paging Query Provider using standard SQL:2003 windowing functions.  These features are supported by
+ * DB2, Oracle, SQL Server 2005, Sybase and Apache Derby version 10.4.1.3
+ *
+ * @author Thomas Risberg
+ * @since 2.0
+ */
+public class SimpleDelegatingPagingQueryProvider extends AbstractSqlPagingQueryProvider {
+
+	/* List of supported database products */
+	public static final List<String> supportedDatabaseProducts = Arrays.asList(
+			"Apache Derby",
+			"DB2",
+			"HSQL Database Engine",
+			"Microsoft SQL Server",
+			"MySQL",
+			"Oracle",
+			"PostgreSQL",
+			"Sybase"
+		);
+
+	AbstractSqlPagingQueryProvider delegate;
+
+	@Override
+	public void init(DataSource dataSource) throws Exception {
+		super.init(dataSource);
+		String databaseProductName = JdbcUtils.commonDatabaseName(
+				JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductName").toString());
+		if ("Apache Derby".equals(databaseProductName)) {
+			delegate = new DerbyPagingQueryProvider();
+		}
+		else if ("DB2".equals(databaseProductName)) {
+			delegate = new Db2PagingQueryProvider();
+		}
+		else if ("HSQL Database Engine".equals(databaseProductName)) {
+			delegate = new HsqlPagingQueryProvider();
+		}
+		else if ("Microsoft SQL Server".equals(databaseProductName)) {
+			delegate = new SqlServerPagingQueryProvider();
+		}
+		else if ("MySQL".equals(databaseProductName)) {
+			delegate = new MySqlPagingQueryProvider();
+		}
+		else if ("Oracle".equals(databaseProductName)) {
+			delegate = new OraclePagingQueryProvider();
+		}
+		else if ("PostgreSQL".equals(databaseProductName)) {
+			delegate = new PostgresPagingQueryProvider();
+		}
+		else if ("Sybase".equals(databaseProductName)) {
+			delegate = new SybasePagingQueryProvider();
+		}
+		else {
+			throw new InvalidDataAccessResourceUsageException(databaseProductName +
+					" is not a supported database.  The supported databases are " +
+					supportedDatabaseProducts.toString());
+		}
+		delegate.setSelectClause(this.getSelectClause());
+		delegate.setFromClause(this.getFromClause());
+		if (this.getWhereClause() != null) {
+			delegate.setWhereClause(this.getWhereClause());
+		}
+		delegate.setSortKey(this.getSortKey());
+		delegate.init(dataSource);
+	}
+
+	@Override
+	public String generateFirstPageQuery(int pageSize) {
+		return delegate.generateFirstPageQuery(pageSize);
+	}
+
+	@Override
+	public String generateRemainingPagesQuery(int pageSize) {
+		return delegate.generateRemainingPagesQuery(pageSize);
+	}
+
+	@Override
+	public String generateJumpToItemQuery(int itemIndex, int pageSize) {
+		return delegate.generateJumpToItemQuery(itemIndex, pageSize);
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlServerPagingQueryProvider.java

@@ -0,0 +1,32 @@
+package org.springframework.batch.item.database.support;
+
+/**
+ * Oracle implementation of a  {@link org.springframework.batch.item.database.support.PagingQueryProvider} using
+ * database specific features.
+ *
+ * @author Thomas Risberg
+ * @since 2.0
+ */
+public class SqlServerPagingQueryProvider extends SqlWindowingPagingQueryProvider {
+
+	@Override
+	public String generateFirstPageQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append("TOP ").append(pageSize).append(" ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(getWhereClause() == null ? "" : " WHERE " + getWhereClause());
+
+		return sql.toString();
+	}
+
+	@Override
+	public String generateRemainingPagesQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append("TOP ").append(pageSize).append(" ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(" WHERE ").append(getSortKey()).append(" > ?");
+		sql.append(getWhereClause() == null ? "" : " AND " + getWhereClause());
+
+		return sql.toString();
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SqlWindowingPagingQueryProvider.java

@@ -24,6 +24,7 @@ package org.springframework.batch.item.database.support;
  */
 public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvider {
 
+	@Override
 	public String generateFirstPageQuery(int pageSize) {
 		StringBuilder sql = new StringBuilder();
 		sql.append("SELECT * FROM ( ");
@@ -35,6 +36,7 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 		return sql.toString();
 	}
 
+	@Override
 	public String generateRemainingPagesQuery(int pageSize) {
 		StringBuilder sql = new StringBuilder();
 		sql.append("SELECT * FROM ( ");
@@ -47,6 +49,7 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 		return sql.toString();
 	}
 
+	@Override
 	public String generateJumpToItemQuery(int itemIndex, int pageSize) {
 		int page = itemIndex / pageSize;
 		int lastRowNum = (page * pageSize);

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SybasePagingQueryProvider.java

@@ -0,0 +1,32 @@
+package org.springframework.batch.item.database.support;
+
+/**
+ * Oracle implementation of a  {@link PagingQueryProvider} using
+ * database specific features.
+ *
+ * @author Thomas Risberg
+ * @since 2.0
+ */
+public class SybasePagingQueryProvider extends SqlWindowingPagingQueryProvider {
+
+	@Override
+	public String generateFirstPageQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append("TOP ").append(pageSize).append(" ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(getWhereClause() == null ? "" : " WHERE " + getWhereClause());
+
+		return sql.toString();
+	}
+
+	@Override
+	public String generateRemainingPagesQuery(int pageSize) {
+		StringBuilder sql = new StringBuilder();
+		sql.append("SELECT ").append("TOP ").append(pageSize).append(" ").append(getSelectClause());
+		sql.append(" FROM ").append(getFromClause());
+		sql.append(" WHERE ").append(getSortKey()).append(" > ?");
+		sql.append(getWhereClause() == null ? "" : " AND " + getWhereClause());
+
+		return sql.toString();
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/Db2PagingQueryProviderTests.java

@@ -0,0 +1,36 @@
+package org.springframework.batch.item.database.support;
+
+import org.junit.Test;
+import org.junit.Assert;
+
+/**
+ * @author Thomas Risberg
+ */
+public class Db2PagingQueryProviderTests extends AbstractSqlPagingQueryProviderTests {
+
+	public Db2PagingQueryProviderTests() {
+		pagingQueryProvider = new Db2PagingQueryProvider();
+	}
+
+	@Test
+	@Override
+	public void testGenerateFirstPageQuery() {
+		String sql = "SELECT id, name, age FROM foo WHERE bar = 1 FETCH FIRST 100 ROWS ONLY";
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateRemainingPagesQuery() {
+		String sql = "SELECT id, name, age FROM foo WHERE id > ? AND bar = 1 FETCH FIRST 100 ROWS ONLY";
+		String s = pagingQueryProvider.generateRemainingPagesQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateJumpToItemQuery() {
+		String sql = "SELECT SORT_KEY FROM ( SELECT id AS SORT_KEY, ROW_NUMBER() OVER (ORDER BY id ASC) AS ROW_NUMBER FROM foo WHERE bar = 1) WHERE ROW_NUMBER = 100";
+		String s = pagingQueryProvider.generateJumpToItemQuery(145, pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/OraclePagingQueryProviderTests.java

@@ -0,0 +1,36 @@
+package org.springframework.batch.item.database.support;
+
+import org.junit.Test;
+import org.junit.Assert;
+
+/**
+ * @author Thomas Risberg
+ */
+public class OraclePagingQueryProviderTests extends AbstractSqlPagingQueryProviderTests {
+
+	public OraclePagingQueryProviderTests() {
+		pagingQueryProvider = new OraclePagingQueryProvider();
+	}
+
+	@Test
+	@Override
+	public void testGenerateFirstPageQuery() {
+		String sql = "SELECT id, name, age FROM foo WHERE ROWNUM <= 100 AND bar = 1";
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateRemainingPagesQuery() {
+		String sql = "SELECT id, name, age FROM foo WHERE id > ? AND ROWNUM <= 100 AND bar = 1";
+		String s = pagingQueryProvider.generateRemainingPagesQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateJumpToItemQuery() {
+		String sql = "SELECT SORT_KEY FROM ( SELECT id AS SORT_KEY, ROW_NUMBER() OVER (ORDER BY id ASC) AS ROW_NUMBER FROM foo WHERE bar = 1) WHERE ROW_NUMBER = 100";
+		String s = pagingQueryProvider.generateJumpToItemQuery(145, pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProviderTests.java

@@ -0,0 +1,234 @@
+package org.springframework.batch.item.database.support;
+
+import static org.junit.Assert.fail;
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+
+import org.junit.Test;
+import org.junit.Assert;
+import org.junit.Before;
+import org.easymock.EasyMock;
+import org.springframework.dao.InvalidDataAccessResourceUsageException;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+
+/**
+ * @author Thomas Risberg
+ */
+public class SimpleDelegatingPagingQueryProviderTests {
+
+	protected AbstractSqlPagingQueryProvider pagingQueryProvider;
+	protected int pageSize;
+	DataSource ds;
+	Connection con;
+	DatabaseMetaData dmd;
+
+	@Before
+	public void onSetUp() throws Exception {
+		ds = createMock(DataSource.class);
+		con = createMock(Connection.class);
+		dmd = createMock(DatabaseMetaData.class);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(ds.getConnection()).andReturn(con);
+
+		pagingQueryProvider = new SimpleDelegatingPagingQueryProvider();
+		initializeQueryProvider(pagingQueryProvider);
+	}
+
+	@Test
+	public void testApacheDerby() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new DerbyPagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		expect(dmd.getDatabaseProductName()).andReturn("Apache Derby");
+		DatabaseMetaData dmd2 = createMock(DatabaseMetaData.class);
+		expect(dmd2.getDatabaseProductVersion()).andReturn("10.4.1.3");
+		expect(con.getMetaData()).andReturn(dmd2);
+		expect(ds.getConnection()).andReturn(con);
+
+		EasyMock.replay(dmd2);
+		EasyMock.replay(con);
+		EasyMock.replay(dmd);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+		EasyMock.verify(dmd2);
+	}
+
+	@Test
+	public void testDb2() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new Db2PagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("DB2");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testHsql() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new HsqlPagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("HSQL Database Engine");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testSqlServer() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new SqlServerPagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("Microsoft SQL Server");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testMySql() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new MySqlPagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("MySQL");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testOracle() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new OraclePagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("Oracle");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testPostgres() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new PostgresPagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("PostgreSQL");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testSybase() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new SybasePagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("Sybase");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		pagingQueryProvider.init(ds);
+		String sql = queryProviderToBeUsed.generateFirstPageQuery(pageSize);
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	@Test
+	public void testUnsupportedDatabase() throws Exception {
+
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("MyDB");
+		EasyMock.replay(dmd);
+		EasyMock.replay(con);
+		EasyMock.replay(ds);
+
+		try {
+			pagingQueryProvider.init(ds);
+			fail("Expected an InvalidDataAccessResourceUsageException since the MyDB database is not supported");
+		} catch (InvalidDataAccessResourceUsageException e) {
+			// expected
+		}
+
+		EasyMock.verify(ds);
+		EasyMock.verify(con);
+		EasyMock.verify(dmd);
+	}
+
+	private void initializeQueryProvider(AbstractSqlPagingQueryProvider queryProvider) throws Exception {
+		queryProvider.setSelectClause("id, name, age");
+		queryProvider.setFromClause("foo");
+		queryProvider.setWhereClause("bar = 1");
+		queryProvider.setSortKey("id");
+	}
+
+
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SqlServerPagingQueryProviderTests.java

@@ -0,0 +1,36 @@
+package org.springframework.batch.item.database.support;
+
+import org.junit.Test;
+import org.junit.Assert;
+
+/**
+ * @author Thomas Risberg
+ */
+public class SqlServerPagingQueryProviderTests extends AbstractSqlPagingQueryProviderTests {
+
+	public SqlServerPagingQueryProviderTests() {
+		pagingQueryProvider = new SqlServerPagingQueryProvider();
+	}
+
+	@Test
+	@Override
+	public void testGenerateFirstPageQuery() {
+		String sql = "SELECT TOP 100 id, name, age FROM foo WHERE bar = 1";
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateRemainingPagesQuery() {
+		String sql = "SELECT TOP 100 id, name, age FROM foo WHERE id > ? AND bar = 1";
+		String s = pagingQueryProvider.generateRemainingPagesQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateJumpToItemQuery() {
+		String sql = "SELECT SORT_KEY FROM ( SELECT id AS SORT_KEY, ROW_NUMBER() OVER (ORDER BY id ASC) AS ROW_NUMBER FROM foo WHERE bar = 1) WHERE ROW_NUMBER = 100";
+		String s = pagingQueryProvider.generateJumpToItemQuery(145, pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SybasePagingQueryProviderTests.java

@@ -0,0 +1,36 @@
+package org.springframework.batch.item.database.support;
+
+import org.junit.Test;
+import org.junit.Assert;
+
+/**
+ * @author Thomas Risberg
+ */
+public class SybasePagingQueryProviderTests extends AbstractSqlPagingQueryProviderTests {
+
+	public SybasePagingQueryProviderTests() {
+		pagingQueryProvider = new SybasePagingQueryProvider();
+	}
+
+	@Test
+	@Override
+	public void testGenerateFirstPageQuery() {
+		String sql = "SELECT TOP 100 id, name, age FROM foo WHERE bar = 1";
+		String s = pagingQueryProvider.generateFirstPageQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateRemainingPagesQuery() {
+		String sql = "SELECT TOP 100 id, name, age FROM foo WHERE id > ? AND bar = 1";
+		String s = pagingQueryProvider.generateRemainingPagesQuery(pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+
+	@Test @Override
+	public void testGenerateJumpToItemQuery() {
+		String sql = "SELECT SORT_KEY FROM ( SELECT id AS SORT_KEY, ROW_NUMBER() OVER (ORDER BY id ASC) AS ROW_NUMBER FROM foo WHERE bar = 1) WHERE ROW_NUMBER = 100";
+		String s = pagingQueryProvider.generateJumpToItemQuery(145, pageSize);
+		Assert.assertEquals("", sql, s);
+	}
+}