BATCH-2478: Improve JdbcPagingItemReader ResultSet access

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2013 the original author or authors.
+ * Copyright 2006-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.
@@ -330,11 +330,15 @@ private class PagingRowMapper implements RowMapper<T> {
 		public T mapRow(ResultSet rs, int rowNum) throws SQLException {
 			startAfterValues = new LinkedHashMap<String, Object>();
 			for (Map.Entry<String, Order> sortKey : queryProvider.getSortKeys().entrySet()) {
-				startAfterValues.put(sortKey.getKey(), rs.getObject(sortKey.getKey()));
+				startAfterValues.put(sortKey.getKey(), findObject(rs, sortKey.getKey()));
 			}
 
 			return rowMapper.mapRow(rs, rowNum);
 		}
+
+		private Object findObject(ResultSet rs, String columnName) throws SQLException{
+			return rs.getObject(queryProvider.getSortKeyResultName(columnName));
+		}
 	}
 
 	private JdbcTemplate getJdbcTemplate() {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2012 the original author or authors.
+ * Copyright 2006-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.
@@ -85,7 +85,14 @@ public interface PagingQueryProvider {
 	 * @return the sort keys used to order the query
 	 */
 	Map<String, Order> getSortKeys();
-
+
+	/**
+	 * Gives the name of the specified sort key, as it should be reffered in the context of a {@link java.sql.ResultSet}.
+	 *
+	 * @return the key's name, as it is used in a {@link java.sql.ResultSet}.
+	 */
+	String getSortKeyResultName(String key);
+
 	/**
 	 * Returns either a String to be used as the named placeholder for a sort key value (based on the column name)
 	 * or a ? for unnamed parameters.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2012 the original author or authors.
+ * Copyright 2006-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.
@@ -59,6 +59,8 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi
 
 	private Map<String, Order> sortKeys = new LinkedHashMap<String, Order>();
 
+	private Map<String, String> sortKeyResultNames = new LinkedHashMap<String, String>();
+
 	private String groupClause;
 
 	private int parameterCount;
@@ -248,6 +250,40 @@ private String removeKeyWord(String keyWord, String clause) {
 		}
 	}
 
+	/**
+	 * Set the names of sort keys as they are to be retrieved from a result set.
+	 *
+	 * Keys that are not included will be determined as described in {@link #getSortKeyResultName}.
+	 */
+	public void setSortKeyResultNames(Map<String, String> sortKeyResultNames) {
+		this.sortKeyResultNames = sortKeyResultNames;
+	}
+
+	/**
+	 * Determines the name to be used for looking up the given key in a result set.
+	 *
+	 * First, if the key is set through {@link #setSortKeyResultNames}, the
+	 *	corresponding result will be returned.
+	 * Second, if the key contains a table separator ('.'), then only the portion
+	 *	of the key after the table separator is returned.
+	 * Otherwise, the key is returned as it was given.
+	 *
+	 * @return the name of the key, in the context of a {@link java.sql.ResultSet}.
+	 */
+	@Override
+	public String getSortKeyResultName(String key) {
+		if (sortKeyResultNames.containsKey(key)) {
+			return sortKeyResultNames.get(key);
+		}
+
+		int tableSeparatorIndex = key.lastIndexOf('.');
+		if (tableSeparatorIndex != -1) {
+			return key.substring(tableSeparatorIndex + 1);
+		}
+
+		return key;
+	}
+
 	/**
 	 *
 	 * @return sortKey key to use to sort and limit page content (without alias)

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/AbstractItemReaderTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2009-2010 the original author or authors.
+ * Copyright 2009-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.
@@ -71,8 +71,9 @@ public void testRead() throws Exception {
 	@Test
 	public void testEmptyInput() throws Exception {
 		pointToEmptyInput(tested);
-		tested.read();
-		assertNull(tested.read());
+
+		final Foo nullFoo = tested.read();
+		assertNull(nullFoo);
 	}
 
 	/**

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

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2012-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;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.database.support.HsqlPagingQueryProvider;
+import org.springframework.batch.item.sample.Foo;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.test.context.ContextConfiguration;
+/**
+ * Tests for {@link JpaPagingItemReader} where multiple tables are involved, requiring the sort key to 
+ * be given table-qualified.
+ *
+ * @author Thomas Risberg
+ * @author Michael Minella
+ * @author Bridger Howell
+ */
+@ContextConfiguration(inheritLocations = false)
+public class JdbcPagingItemReaderMultiTableTests extends AbstractGenericDataSourceItemReaderIntegrationTests {
+
+	@Override
+	protected ItemReader<Foo> createItemReader() throws Exception {
+		JdbcPagingItemReader<Foo> inputSource = new JdbcPagingItemReader<Foo>();
+		inputSource.setDataSource(dataSource);
+		HsqlPagingQueryProvider queryProvider = new HsqlPagingQueryProvider();
+		queryProvider.setSelectClause("select T_FOOS.ID, T_FOOS.NAME, T_FOOS.VALUE");
+		queryProvider.setFromClause("from T_FOOS, T_BARS");
+		queryProvider.setWhereClause("where T_BARS.ID = 1");
+		Map<String, Order> sortKeys = new LinkedHashMap<String, Order>();
+		sortKeys.put("T_FOOS.ID", Order.ASCENDING);
+		queryProvider.setSortKeys(sortKeys);
+		inputSource.setQueryProvider(queryProvider);
+		inputSource.setRowMapper(
+				new RowMapper<Foo>() {
+					@Override
+					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;
+	}
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2012 the original author or authors.
+ * Copyright 2006-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.
@@ -103,7 +103,25 @@ public void testGenerateJumpToItemQueryForFirstPageWithMultipleSortKeys() {
 		String s = pagingQueryProvider.generateJumpToItemQuery(45, pageSize);
 		assertEquals(getJumpToItemQueryForFirstPageWithMultipleSortKeys(), s);
 	}
-
+
+	@Test
+	public void testGetSortKeyResultNameWithoutMap() {
+		assertEquals("id", pagingQueryProvider.getSortKeyResultName("id"));
+		assertEquals("column", pagingQueryProvider.getSortKeyResultName("table.column"));
+		assertEquals("column", pagingQueryProvider.getSortKeyResultName("schema.table.column"));
+	}
+
+	@Test
+	public void testGetSortKeyResultNameWithMap() {
+		Map<String, String> sortKeyResultNames = new LinkedHashMap<>();
+		sortKeyResultNames.put("foo.id", "foo_id");
+		pagingQueryProvider.setSortKeyResultNames(sortKeyResultNames);
+
+		assertEquals("foo_id", pagingQueryProvider.getSortKeyResultName("foo.id"));
+		assertEquals("column", pagingQueryProvider.getSortKeyResultName("column"));
+		assertEquals("id", pagingQueryProvider.getSortKeyResultName("bar.id"));
+	}
+
 	@Test
 	public abstract void testGenerateFirstPageQuery();
 

spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/database/JdbcPagingItemReaderMultiTableTests-context.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:util="http://www.springframework.org/schema/util"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
+	http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.1.xsd">
+
+	<bean class="test.jdbc.datasource.DataSourceInitializer">
+		<property name="dataSource" ref="dataSource"/>
+		<property name="initScripts">
+			<list>
+				<value>classpath:org/springframework/batch/item/database/init-foo-schema-hsqldb.sql</value>
+				<value>classpath:org/springframework/batch/item/database/init-bar-schema-hsqldb.sql</value>
+			</list>
+		</property>
+	</bean>
+	<bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
+		<property name="driverClassName" value="org.hsqldb.jdbcDriver" />
+		<property name="url" value="jdbc:hsqldb:mem:testdb" />
+	</bean>
+
+	<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
+		<property name="dataSource" ref="dataSource" />
+	</bean>
+
+	<bean id="xincrementerParent" class="org.springframework.jdbc.support.incrementer.HsqlMaxValueIncrementer"
+		abstract="true">
+		<property name="dataSource" ref="dataSource" />
+		<property name="columnName" value="ID" />
+	</bean>
+</beans>

spring-batch-infrastructure/src/test/resources/org/springframework/batch/item/database/init-bar-schema-hsqldb.sql

@@ -0,0 +1,12 @@
+DROP TABLE T_BARS if exists;
+
+CREATE TABLE T_BARS (
+       ID BIGINT NOT NULL,
+       NAME VARCHAR(45),
+       VALUE INTEGER
+);
+
+ALTER TABLE T_BARS ADD PRIMARY KEY (ID);
+
+INSERT INTO t_bars (id, name, value) VALUES (1, 'baz1', 1);
+