diff --git a/docs/src/site/docbook/reference/execution.xml b/docs/src/site/docbook/reference/execution.xml
index 044ec893b..09df5e0c4 100644
--- a/docs/src/site/docbook/reference/execution.xml
+++ b/docs/src/site/docbook/reference/execution.xml
@@ -457,7 +457,7 @@
   &lt;/bean&gt;</programlisting>
 
         <para>The databaseType property indicates the type of incrementer that
-        must be used. Options include: "db2", "derby", "hsql", "mysql",
+        must be used. Options include: "db2", "db2zos", "derby", "hsql", "mysql",
         "oracle", and "postgres".</para>
       </section>
 
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactory.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactory.java
index d09220625..f1dd121d2 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactory.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactory.java
@@ -32,6 +32,7 @@ import org.springframework.jdbc.support.incrementer.OracleSequenceMaxValueIncrem
 import org.springframework.jdbc.support.incrementer.PostgreSQLSequenceMaxValueIncrementer;
 import org.springframework.jdbc.support.incrementer.SqlServerMaxValueIncrementer;
 import org.springframework.jdbc.support.incrementer.SybaseMaxValueIncrementer;
+import org.springframework.jdbc.support.incrementer.DB2MainframeSequenceMaxValueIncrementer;
 
 /**
  * Default implementation of the {@link DataFieldMaxValueIncrementerFactory}
@@ -41,6 +42,7 @@ import org.springframework.jdbc.support.incrementer.SybaseMaxValueIncrementer;
  * 
  * <ul>
  * <li>db2</li>
+ * <li>db2zos</li>
  * <li>derby</li>
  * <li>hsql</li>
  * <li>mysql</li>
@@ -81,6 +83,9 @@ public class DefaultDataFieldMaxValueIncrementerFactory implements DataFieldMaxV
 		if (databaseType == DB2) {
 			return new DB2SequenceMaxValueIncrementer(dataSource, incrementerName);
 		}
+		else if (databaseType == DB2ZOS) {
+			return new DB2MainframeSequenceMaxValueIncrementer(dataSource, incrementerName);
+		}
 		else if (databaseType == DERBY) {
 			return new DerbyMaxValueIncrementer(dataSource, incrementerName, incrementerColumnName);
 		}
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProvider.java
index 3f979f917..e6b6ba853 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProvider.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProvider.java
@@ -41,7 +41,7 @@ public class SimpleDelegatingPagingQueryProvider extends AbstractSqlPagingQueryP
 		if (type == DERBY) {
 			delegate = new DerbyPagingQueryProvider();
 		}
-		else if (type == DB2) {
+		else if (type == DB2 || type == DB2ZOS) {
 			delegate = new Db2PagingQueryProvider();
 		}
 		else if (type == HSQL) {
diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/DatabaseType.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/DatabaseType.java
index 6c1c33918..df55774bf 100644
--- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/DatabaseType.java
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/support/DatabaseType.java
@@ -24,7 +24,8 @@ public enum DatabaseType {
 
 	DERBY("Apache Derby"), 
 	DB2("DB2"), 
-	HSQL("HSQL Database Engine"), 
+	DB2ZOS("DB2ZOS"), 
+	HSQL("HSQL Database Engine"),
 	SQLSERVER("Microsoft SQL Server"),
 	MYSQL("MySQL"),
 	ORACLE("Oracle"),
@@ -76,9 +77,21 @@ public enum DatabaseType {
 	 * @throws MetaDataAccessException
 	 */
 	public static DatabaseType fromMetaData(DataSource dataSource) throws MetaDataAccessException{
-		String databaseProductName = JdbcUtils.commonDatabaseName(
-				JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductName").toString());
-		
+		String databaseProductName =
+				JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductName").toString();
+		if ("DB2".equals(databaseProductName)) {
+			String databaseProductVersion =
+					JdbcUtils.extractDatabaseMetaData(dataSource, "getDatabaseProductVersion").toString();
+			if (!databaseProductVersion.startsWith("SQL")) {
+				databaseProductName = "DB2ZOS";
+			}
+			else {
+				databaseProductName = JdbcUtils.commonDatabaseName(databaseProductName);
+			}
+		}
+		else {
+			databaseProductName = JdbcUtils.commonDatabaseName(databaseProductName);
+		}
 		return fromProductName(databaseProductName);
 	}
 }
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactoryTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactoryTests.java
index 7032ca170..3eb51f1bf 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactoryTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/DefaultDataFieldMaxValueIncrementerFactoryTests.java
@@ -28,6 +28,7 @@ import org.springframework.jdbc.support.incrementer.OracleSequenceMaxValueIncrem
 import org.springframework.jdbc.support.incrementer.PostgreSQLSequenceMaxValueIncrementer;
 import org.springframework.jdbc.support.incrementer.SqlServerMaxValueIncrementer;
 import org.springframework.jdbc.support.incrementer.SybaseMaxValueIncrementer;
+import org.springframework.jdbc.support.incrementer.DB2MainframeSequenceMaxValueIncrementer;
 
 /**
  * @author Lucas Ward
@@ -49,6 +50,7 @@ public class DefaultDataFieldMaxValueIncrementerFactoryTests extends TestCase {
 	
 	public void testSupportedDatabaseType(){
 		assertTrue(factory.isSupportedIncrementerType("db2"));
+		assertTrue(factory.isSupportedIncrementerType("db2zos"));
 		assertTrue(factory.isSupportedIncrementerType("mysql"));
 		assertTrue(factory.isSupportedIncrementerType("derby"));
 		assertTrue(factory.isSupportedIncrementerType("oracle"));
@@ -86,6 +88,10 @@ public class DefaultDataFieldMaxValueIncrementerFactoryTests extends TestCase {
 		assertTrue(factory.getIncrementer("db2", "NAME") instanceof DB2SequenceMaxValueIncrementer);
 	}
 	
+	public void testDb2zos(){
+		assertTrue(factory.getIncrementer("db2zos", "NAME") instanceof DB2MainframeSequenceMaxValueIncrementer);
+	}
+
 	public void testMysql(){
 		assertTrue(factory.getIncrementer("mysql", "NAME") instanceof MySQLMaxValueIncrementer);
 	}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProviderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProviderTests.java
index 421a6495a..6f6338365 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProviderTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/support/SimpleDelegatingPagingQueryProviderTests.java
@@ -67,7 +67,30 @@ public class SimpleDelegatingPagingQueryProviderTests {
 		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new Db2PagingQueryProvider();
 		initializeQueryProvider(queryProviderToBeUsed);
 
+		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("DB2/Linux");
+		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 testDb2ZOS() throws Exception {
+		AbstractSqlPagingQueryProvider queryProviderToBeUsed = new Db2PagingQueryProvider();
+		initializeQueryProvider(queryProviderToBeUsed);
+
 		EasyMock.expect(dmd.getDatabaseProductName()).andReturn("DB2");
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(ds.getConnection()).andReturn(con);
+		EasyMock.expect(dmd.getDatabaseProductVersion()).andReturn("DSN08015");
 		EasyMock.replay(dmd);
 		EasyMock.replay(con);
 		EasyMock.replay(ds);
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/DatabaseTypeTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/DatabaseTypeTests.java
index 9dae0a6f1..1f5d937ea 100644
--- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/DatabaseTypeTests.java
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/support/DatabaseTypeTests.java
@@ -22,6 +22,7 @@ public class DatabaseTypeTests {
 	public void testFromProductName(){
 		assertEquals(DERBY, fromProductName("Apache Derby"));
 		assertEquals(DB2, fromProductName("DB2"));
+		assertEquals(DB2ZOS, fromProductName("DB2ZOS"));
 		assertEquals(HSQL, fromProductName("HSQL Database Engine"));
 		assertEquals(SQLSERVER, fromProductName("Microsoft SQL Server"));
 		assertEquals(MYSQL, fromProductName("MySQL"));
@@ -37,8 +38,107 @@ public class DatabaseTypeTests {
 	}
 	
 	@Test
-	public void testFromMetaData() throws Exception{
+	public void testFromMetaDataForDerby() throws Exception{
 		
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("Apache Derby");
+		replay(dmd,ds,con);
+		
+		assertEquals(DERBY, DatabaseType.fromMetaData(ds));
+		
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForDB2() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("DB2/Linux");
+		replay(dmd,ds,con);
+
+		assertEquals(DB2, DatabaseType.fromMetaData(ds));
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForDB2ZOS() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("DB2");
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductVersion()).andReturn("DSN08015");
+		replay(dmd,ds,con);
+
+		assertEquals(DB2ZOS, DatabaseType.fromMetaData(ds));
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForHsql() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("HSQL Database Engine");
+		replay(dmd,ds,con);
+
+		assertEquals(HSQL, DatabaseType.fromMetaData(ds));
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForSqlServer() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("Microsoft SQL Server");
+		replay(dmd,ds,con);
+
+		assertEquals(SQLSERVER, DatabaseType.fromMetaData(ds));
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForMySql() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("MySQL");
+		replay(dmd,ds,con);
+
+		assertEquals(MYSQL, DatabaseType.fromMetaData(ds));
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForOracle() throws Exception{
+
 		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
 		DataSource ds = createMock(DataSource.class);
 		Connection con = createMock(Connection.class);
@@ -46,9 +146,41 @@ public class DatabaseTypeTests {
 		expect(con.getMetaData()).andReturn(dmd);
 		expect(dmd.getDatabaseProductName()).andReturn("Oracle");
 		replay(dmd,ds,con);
-		
+
 		assertEquals(ORACLE, DatabaseType.fromMetaData(ds));
-		
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForPostgres() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("PostgreSQL");
+		replay(dmd,ds,con);
+
+		assertEquals(POSTGRES, DatabaseType.fromMetaData(ds));
+
+		verify(dmd,ds,con);
+	}
+
+	@Test
+	public void testFromMetaDataForSybase() throws Exception{
+
+		DatabaseMetaData dmd = createMock(DatabaseMetaData.class);
+		DataSource ds = createMock(DataSource.class);
+		Connection con = createMock(Connection.class);
+		expect(ds.getConnection()).andReturn(con);
+		expect(con.getMetaData()).andReturn(dmd);
+		expect(dmd.getDatabaseProductName()).andReturn("Adaptive Server Enterprise");
+		replay(dmd,ds,con);
+
+		assertEquals(SYBASE, DatabaseType.fromMetaData(ds));
+
 		verify(dmd,ds,con);
 	}
 }