BATCH-2040: Fix for erroneous SQL-Query in PagingQueryProvider

- added method for replacing qualifiers with subquery-qualifiers in
select clause
- refactored buildsort method
- Replaced HashMap with LinkedHashMap for right sort orders

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

@@ -246,10 +246,20 @@ public class SqlPagingQueryUtils {
 	 * @return a String that can be appended to an ORDER BY clause.
 	 */
 	public static String buildSortClause(AbstractSqlPagingQueryProvider provider) {
+		return buildSortClause(provider.getSortKeys());
+	}
+	
+	/**
+	 * Generates ORDER BY attributes based on the sort keys.
+	 * 
+	 * @param sortKeys
+	 * @return a String that can be appended to an ORDER BY clause.
+	 */
+	public static String buildSortClause(Map<String, Order> sortKeys) {
 		StringBuilder builder = new StringBuilder();
 		String prefix = "";
 		
-		for (Map.Entry<String, Order> sortKey : provider.getSortKeys().entrySet()) {
+		for (Map.Entry<String, Order> sortKey : sortKeys.entrySet()) {
 			builder.append(prefix);
 			
 			prefix = ", ";

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

@@ -16,6 +16,7 @@
 
 package org.springframework.batch.item.database.support;
 
+import java.util.LinkedHashMap;
 import java.util.Map;
 
 import org.springframework.batch.item.database.Order;
@@ -102,7 +103,7 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 
 		StringBuilder sql = new StringBuilder();
 		sql.append("SELECT ");
-		buildSortKeySelect(sql);
+		buildSortKeySelect(sql, getSortKeysReplaced(extractTableAlias()));
 		sql.append(" FROM ( ");
 		sql.append("SELECT ");
 		buildSortKeySelect(sql);
@@ -115,14 +116,30 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 		sql.append(getOverSubstituteClauseEnd());
 		sql.append(") ").append(getSubQueryAlias()).append("WHERE ").append(extractTableAlias()).append(
 				"ROW_NUMBER = ").append(lastRowNum);
-		sql.append(" ORDER BY ").append(SqlPagingQueryUtils.buildSortClause(this));
+		sql.append(" ORDER BY ").append(SqlPagingQueryUtils.buildSortClause(getSortKeysReplaced(extractTableAlias())));
 
 		return sql.toString();
 	}
 
-	private void buildSortKeySelect(StringBuilder sql) {
-		String prefix = "";
+	private Map<String, Order> getSortKeysReplaced(Object qualifierReplacement) {
+		final String newQualifier = "" + qualifierReplacement;
+		final Map<String, Order> sortKeys = new LinkedHashMap<String, Order>();
 		for (Map.Entry<String, Order> sortKey : getSortKeys().entrySet()) {
+			sortKeys.put(sortKey.getKey().replaceFirst("^.*\\.", newQualifier), sortKey.getValue());
+		}
+		return sortKeys;
+	}
+	
+	private void buildSortKeySelect(StringBuilder sql) {
+		buildSortKeySelect(sql, null);
+	}
+	
+	private void buildSortKeySelect(StringBuilder sql, Map<String, Order> sortKeys) {
+		String prefix = "";
+		if (sortKeys == null) {
+			sortKeys = getSortKeys();
+		}
+		for (Map.Entry<String, Order> sortKey : sortKeys.entrySet()) {
 			sql.append(prefix);
 			prefix = ", ";
 			sql.append(sortKey.getKey());

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

@@ -17,8 +17,12 @@ package org.springframework.batch.item.database.support;
 
 import static org.junit.Assert.assertEquals;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.junit.Assert;
 import org.junit.Test;
+import org.springframework.batch.item.database.Order;
 
 /**
  * @author Thomas Risberg
@@ -97,6 +101,20 @@ public class SqlWindowingPagingQueryProviderTests extends AbstractSqlPagingQuery
 		String s = pagingQueryProvider.generateJumpToItemQuery(45, pageSize);
 		assertEquals(sql, s);
 	}
+	
+	@Test
+	public void testGenerateJumpToItemQueryForTableQualifierReplacement() {
+		pagingQueryProvider.setFromClause("foo_e E, foo_i I");
+		pagingQueryProvider.setWhereClause("E.id=I.id");
+		
+		Map<String, Order> sortKeys = new HashMap<String, Order>();
+		sortKeys.put("E.id", Order.DESCENDING);		
+		pagingQueryProvider.setSortKeys(sortKeys);
+		
+		String sql="SELECT TMP_SUB.id FROM ( SELECT E.id, ROW_NUMBER() OVER ( ORDER BY E.id DESC) AS ROW_NUMBER FROM foo_e E, foo_i I WHERE E.id=I.id) AS TMP_SUB WHERE TMP_SUB.ROW_NUMBER = 1 ORDER BY TMP_SUB.id DESC";
+		String s = pagingQueryProvider.generateJumpToItemQuery(45, pageSize);
+		assertEquals(sql, s);
+	}
 
 	@Override
 	public String getFirstPageSqlWithMultipleSortKeys() {