BATCH-1749: JdbcPagingItemReader now supports multi-column keys

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

@@ -37,7 +37,6 @@ import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
-import org.springframework.util.StringUtils;
 
 /**
  * <p>
@@ -87,13 +86,14 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 
 	private NamedParameterJdbcTemplate namedParameterJdbcTemplate;
 
+	@SuppressWarnings("rawtypes")
 	private RowMapper rowMapper;
 
 	private String firstPageSql;
 
 	private String remainingPagesSql;
 
-	private Object startAfterValue;
+	private Map<String, Object> startAfterValues;
 
 	private int fetchSize = VALUE_NOT_SET;
 
@@ -137,6 +137,7 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 	 * {@link org.springframework.jdbc.core.simple.ParameterizedRowMapper}
 	 * implementation
 	 */
+	@SuppressWarnings("rawtypes")
 	public void setRowMapper(RowMapper rowMapper) {
 		this.rowMapper = rowMapper;
 	}
@@ -176,8 +177,8 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	protected void doReadPage() {
-
 		if (results == null) {
 			results = new CopyOnWriteArrayList<T>();
 		}
@@ -214,43 +215,48 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 			}
 			if (this.queryProvider.isUsingNamedParameters()) {
 				query = namedParameterJdbcTemplate.query(remainingPagesSql,
-						getParameterMap(parameterValues, startAfterValue), rowCallback);
+						getParameterMap(parameterValues, startAfterValues), rowCallback);
 			}
 			else {
 				query = getJdbcTemplate().query(remainingPagesSql,
-						getParameterList(parameterValues, startAfterValue).toArray(), rowCallback);
+						getParameterList(parameterValues, startAfterValues).toArray(), rowCallback);
 			}
 		}
 		
-		@SuppressWarnings("unchecked")
 		Collection<T> result = (Collection<T>) query;
 		results.addAll(result);
-
 	}
 
 	@Override
 	public void update(ExecutionContext executionContext) throws ItemStreamException {
 		super.update(executionContext);
-		if (isSaveState() && startAfterValue != null) {
-			executionContext.put(getExecutionContextUserSupport().getKey(START_AFTER_VALUE), startAfterValue);
+		if (isSaveState() && startAfterValues != null) {
+			executionContext.put(getExecutionContextUserSupport().getKey(START_AFTER_VALUE), startAfterValues);
 		}
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void open(ExecutionContext executionContext) {
 		if (isSaveState()) {
-			startAfterValue = executionContext.get(getExecutionContextUserSupport().getKey(START_AFTER_VALUE));
+			startAfterValues = (Map<String, Object>) executionContext.get(getExecutionContextUserSupport().getKey(START_AFTER_VALUE));
+			
+			if(startAfterValues == null) {
+				startAfterValues = new LinkedHashMap<String, Object>();
+			}
 		}
+		
 		super.open(executionContext);
 	}
 
 	@Override
+	@SuppressWarnings({"unchecked", "rawtypes"})
 	protected void doJumpToPage(int itemIndex) {
 		/*
 		 * Normally this would be false (the startAfterValue is enough
 		 * information to restart from.
 		 */
-		if (startAfterValue == null && getPage() > 0) {
+		if (startAfterValues == null && getPage() > 0) {
 
 			String jumpToItemSql;
 			jumpToItemSql = queryProvider.generateJumpToItemQuery(itemIndex, getPageSize());
@@ -265,24 +271,25 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 				}
 			};
 			if (this.queryProvider.isUsingNamedParameters()) {
-				startAfterValue = namedParameterJdbcTemplate.queryForObject(jumpToItemSql,
-						getParameterMap(parameterValues, startAfterValue), startMapper);
+				startAfterValues = (Map<String, Object>) namedParameterJdbcTemplate.queryForObject(jumpToItemSql,
+						getParameterMap(parameterValues, startAfterValues), startMapper);
 			}
 			else {
-				startAfterValue = getJdbcTemplate().queryForObject(jumpToItemSql,
-						getParameterList(parameterValues, startAfterValue).toArray(), startMapper);
+				startAfterValues = (Map<String, Object>) getJdbcTemplate().queryForObject(jumpToItemSql,
+						getParameterList(parameterValues, startAfterValues).toArray(), startMapper);
 			}
-
 		}
 	}
 
-	private Map<String, Object> getParameterMap(Map<String, Object> values, Object sortKeyValue) {
+	private Map<String, Object> getParameterMap(Map<String, Object> values, Map<String, Object> sortKeyValues) {
 		Map<String, Object> parameterMap = new LinkedHashMap<String, Object>();
 		if (values != null) {
 			parameterMap.putAll(values);
 		}
-		if (sortKeyValue != null) {
-			parameterMap.put("_sortKey", sortKeyValue);
+		if (sortKeyValues != null && !sortKeyValues.isEmpty()) {
+			for (Map.Entry<String, Object> sortKey : sortKeyValues.entrySet()) {
+				parameterMap.put("_" + sortKey.getKey(), sortKey.getValue());
+			}
 		}
 		if (logger.isDebugEnabled()) {
 			logger.debug("Using parameterMap:" + parameterMap);
@@ -290,29 +297,37 @@ public class JdbcPagingItemReader<T> extends AbstractPagingItemReader<T> impleme
 		return parameterMap;
 	}
 
-	private List<Object> getParameterList(Map<String, Object> values, Object sortKeyValue) {
+	private List<Object> getParameterList(Map<String, Object> values, Map<String, Object> sortKeyValue) {
 		SortedMap<String, Object> sm = new TreeMap<String, Object>();
 		if (values != null) {
 			sm.putAll(values);
 		}
 		List<Object> parameterList = new ArrayList<Object>();
 		parameterList.addAll(sm.values());
-		if (sortKeyValue != null) {
-			parameterList.add(sortKeyValue);
+		if (sortKeyValue != null && sortKeyValue.size() > 0) {
+			List<Map.Entry<String, Object>> keys = new ArrayList<Map.Entry<String,Object>>(sortKeyValue.entrySet());
+
+			for(int i = 0; i < keys.size(); i++) {
+				for(int j = 0; j < i; j++) {
+					parameterList.add(keys.get(j).getValue());
+				}
+
+				parameterList.add(keys.get(i).getValue());
+			}
 		}
+		
 		if (logger.isDebugEnabled()) {
 			logger.debug("Using parameterList:" + parameterList);
 		}
 		return parameterList;
 	}
 
+	@SuppressWarnings("rawtypes")
 	private class PagingRowMapper implements RowMapper {
 		public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
-			if(StringUtils.hasText(queryProvider.getSortKey())) {
-				startAfterValue = rs.getObject(queryProvider.getSortKey());
-			}
-			else {
-				startAfterValue = rs.getObject(queryProvider.getSortKeyWithoutAlias());
+			startAfterValues = new LinkedHashMap<String, Object>();
+			for (Map.Entry<String, Boolean> sortKey : queryProvider.getSortKeys().entrySet()) {
+				startAfterValues.put(sortKey.getKey(), rs.getObject(sortKey.getKey()));
 			}
 			
 			return rowMapper.mapRow(rs, rowNum);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2008 the original author or authors.
+ * Copyright 2006-2012 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.batch.item.database;
 
+import java.util.Map;
+
 import javax.sql.DataSource;
 
 /**
@@ -23,6 +25,7 @@ import javax.sql.DataSource;
  * Item Readers.
  *
  * @author Thomas Risberg
+ * @author Michael Minella
  * @since 2.0
  */
 public interface PagingQueryProvider {
@@ -76,17 +79,19 @@ public interface PagingQueryProvider {
 	boolean isUsingNamedParameters();
 
 	/**
-	 * The sort key (unique single column name).
+	 * The sort keys.  A Map of the columns that make up the key and a Boolean indicating ascending or descending 
+	 * (ascending = true). 
 	 *  
-	 * @return the sort key used to order the query
+	 * @return the sort keys used to order the query
 	 */
-	String getSortKey();
-
+	Map<String, Boolean> getSortKeys();
+	
 	/**
-	 * The sort key (unique single column name) without alias.
-	 *
-	 * @return the sort key used to order the query (without alias)
+	 * 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.
+	 * 
+	 * @param The sort key name
+	 * @return The string to be used for a parameterized query.
 	 */
-	String getSortKeyWithoutAlias();
-
+	String getSortKeyPlaceHolder(String keyName);
 }

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

@@ -16,16 +16,18 @@
 
 package org.springframework.batch.item.database.support;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
 import javax.sql.DataSource;
 
-import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 import org.springframework.batch.item.database.JdbcParameterUtils;
 import org.springframework.batch.item.database.PagingQueryProvider;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
-
-import java.util.List;
-import java.util.ArrayList;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * Abstract SQL Paging Query Provider to serve as a base class for all provided
@@ -39,7 +41,8 @@ import java.util.ArrayList;
  * 
  * Provides properties and preparation for the mandatory "selectClause" and
  * "fromClause" as well as for the optional "whereClause". Also provides
- * property for the mandatory "sortKey".
+ * property for the mandatory "sortKeys".  <b>Note:</b> The columns that make up 
+ * the sort key must be a true key and not just a column to order by.
  * 
  * @author Thomas Risberg
  * @author Dave Syer
@@ -53,12 +56,10 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi
 	private String fromClause;
 
 	private String whereClause;
-
-	private String sortKey;
 	
-	private String groupClause;
+	private Map<String, Boolean> sortKeys = new TreeMap<String, Boolean>();
 
-	private boolean ascending = true;
+	private String groupClause;
 
 	private int parameterCount;
 
@@ -133,53 +134,19 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi
 	}
 
 	/**
-	 * @param sortKey key to use to sort and limit page content
+	 * @param sortKeys key to use to sort and limit page content
 	 */
-	public void setSortKey(String sortKey) {
-		this.sortKey = sortKey;
+	public void setSortKeys(Map<String, Boolean> sortKeys) {
+		this.sortKeys = sortKeys;
 	}
 
 	/**
-	 * Set the flag that signals that the sort key is applied ascending (default
-	 * true).
+	 * A Map<String, Boolean> of sort columns as the key and boolean for ascending/descending (assending = true).
 	 * 
-	 * @param ascending the ascending value to set
-	 */
-	public void setAscending(boolean ascending) {
-		this.ascending = ascending;
-	}
-
-	/**
-	 * Get the flag that signals that the sort key is applied ascending.
-	 * 
-	 * @return the ascending flag
-	 */
-	public boolean isAscending() {
-		return ascending;
-	}
-
-	/**
-	 *
 	 * @return sortKey key to use to sort and limit page content
 	 */
-	public String getSortKey() {
-		return sortKey;
-	}
-
-	/**
-	 *
-	 * @return sortKey key to use to sort and limit page content (without alias)
-	 */
-	public String getSortKeyWithoutAlias() {
-		String sortKey = getSortKey();
-		int separator = sortKey.indexOf('.');
-		if (separator > 0) {
-			int columnIndex = separator + 1;
-			if (columnIndex < sortKey.length()) {
-				sortKey = sortKey.substring(columnIndex);
-			}
-		}
-		return sortKey;
+	public Map<String, Boolean> getSortKeys() {
+		return sortKeys;
 	}
 
 	public int getParameterCount() {
@@ -196,8 +163,8 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi
 	 * 
 	 * @return place holder for sortKey.
 	 */
-	protected String getSortKeyPlaceHolder() {
-		return usingNamedParameters ? ":_sortKey" : "?";
+	public String getSortKeyPlaceHolder(String keyName) {
+		return usingNamedParameters ? ":_" + keyName : "?";
 	}
 
 	/**
@@ -208,7 +175,7 @@ public abstract class AbstractSqlPagingQueryProvider implements PagingQueryProvi
 		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.notEmpty(sortKeys, "sortKey must be specified");
 		StringBuilder sql = new StringBuilder();
 		sql.append("SELECT ").append(selectClause);
 		sql.append(" FROM ").append(fromClause);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2008 the original author or authors.
+ * Copyright 2006-2012 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.
@@ -16,12 +16,15 @@
 
 package org.springframework.batch.item.database.support;
 
+import java.util.Map;
+
 /**
  * Oracle implementation of a
  * {@link org.springframework.batch.item.database.PagingQueryProvider} using
  * database specific features.
  * 
  * @author Thomas Risberg
+ * @author Michael Minella
  * @since 2.0
  */
 public class OraclePagingQueryProvider extends AbstractSqlPagingQueryProvider {
@@ -42,9 +45,23 @@ public class OraclePagingQueryProvider extends AbstractSqlPagingQueryProvider {
 		int page = itemIndex / pageSize;
 		int offset = (page * pageSize);
 		offset = offset == 0 ? 1 : offset;
-		return SqlPagingQueryUtils.generateRowNumSqlQueryWithNesting(this, this.getSortKey() + " AS SORT_KEY", "SORT_KEY", false, "TMP_ROW_NUM = "
+		String sortKeySelect = this.getSortKeySelect();
+		return SqlPagingQueryUtils.generateRowNumSqlQueryWithNesting(this, sortKeySelect, sortKeySelect, false, "TMP_ROW_NUM = "
 				+ offset);
 	}
+	
+	private String getSortKeySelect() {
+		StringBuilder sql = new StringBuilder();
+		String prefix = "";
+		
+		for (Map.Entry<String, Boolean> sortKey : this.getSortKeys().entrySet()) {
+			sql.append(prefix);
+			prefix = ", ";
+			sql.append(sortKey.getKey());
+		}
+		
+		return sql.toString();
+	}
 
 	private String buildRowNumClause(int pageSize) {
 		return new StringBuilder().append("ROWNUM <= ").append(pageSize).toString();

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

@@ -60,7 +60,7 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean {
 	
 	private String groupClause;
 
-	private String sortKey;
+	private Map<String, Boolean> sortKeys;
 
 	private boolean ascending = true;
 
@@ -125,8 +125,8 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean {
 	/**
 	 * @param sortKey the sortKey to set
 	 */
-	public void setSortKey(String sortKey) {
-		this.sortKey = sortKey;
+	public void setSortKeys(Map<String, Boolean> sortKeys) {
+		this.sortKeys = sortKeys;
 	}
 
 	/**
@@ -159,8 +159,7 @@ public class SqlPagingQueryProviderFactoryBean implements FactoryBean {
 
 		provider.setFromClause(fromClause);
 		provider.setWhereClause(whereClause);
-		provider.setSortKey(sortKey);
-		provider.setAscending(ascending);
+		provider.setSortKeys(sortKeys);
 		if (StringUtils.hasText(selectClause)) {
 			provider.setSelectClause(selectClause);
 		}

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

@@ -16,6 +16,11 @@
 
 package org.springframework.batch.item.database.support;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
 import org.springframework.util.StringUtils;
 
 /**
@@ -46,8 +51,7 @@ public class SqlPagingQueryUtils {
 		sql.append(" FROM ").append(provider.getFromClause());
 		buildWhereClause(provider, remainingPageQuery, sql);
 		buildGroupByClause(provider, sql);
-		sql.append(" ORDER BY ").append(provider.getSortKeyWithoutAlias());
-		buildAscendingClause(provider, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
 		sql.append(" " + limitClause);
 
 		return sql.toString();
@@ -70,8 +74,7 @@ public class SqlPagingQueryUtils {
 		sql.append(" FROM ").append(provider.getFromClause());
 		buildWhereClause(provider, remainingPageQuery, sql);
 		buildGroupByClause(provider, sql);
-		sql.append(" ORDER BY ").append(provider.getSortKeyWithoutAlias());
-		buildAscendingClause(provider, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
 
 		return sql.toString();
 	}
@@ -110,8 +113,7 @@ public class SqlPagingQueryUtils {
 		sql.append(" FROM ").append(provider.getFromClause());
 		buildWhereClause(provider, remainingPageQuery, sql);
 		buildGroupByClause(provider, sql);
-		sql.append(" ORDER BY ").append(provider.getSortKeyWithoutAlias());
-		buildAscendingClause(provider, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
 		sql.append(") WHERE ").append(rowNumClause);
 
 		return sql.toString();
@@ -132,8 +134,7 @@ public class SqlPagingQueryUtils {
 		sql.append(" FROM (SELECT ").append(innerSelectClause).append(" FROM ").append(provider.getFromClause());
 		buildWhereClause(provider, remainingPageQuery, sql);
 		buildGroupByClause(provider, sql);
-		sql.append(" ORDER BY ").append(provider.getSortKeyWithoutAlias());
-		buildAscendingClause(provider, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
 		sql.append(")) WHERE ").append(rowNumClause);
 
 		return sql.toString();
@@ -150,12 +151,11 @@ public class SqlPagingQueryUtils {
 	 */
 	public static String generateLimitJumpToQuery(AbstractSqlPagingQueryProvider provider, String limitClause) {
 		StringBuilder sql = new StringBuilder();
-		sql.append("SELECT ").append(provider.getSortKey()).append(" AS SORT_KEY");
+		sql.append("SELECT ").append(buildSortKeySelect(provider));
 		sql.append(" FROM ").append(provider.getFromClause());
 		sql.append(provider.getWhereClause() == null ? "" : " WHERE " + provider.getWhereClause());
 		buildGroupByClause(provider, sql);
-		sql.append(" ORDER BY ").append(provider.getSortKeyWithoutAlias());
-		buildAscendingClause(provider, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
 		sql.append(" " + limitClause);
 
 		return sql.toString();
@@ -171,23 +171,111 @@ public class SqlPagingQueryUtils {
 	 */
 	public static String generateTopJumpToQuery(AbstractSqlPagingQueryProvider provider, String topClause) {
 		StringBuilder sql = new StringBuilder();
-		sql.append("SELECT ").append(topClause).append(" ").append(provider.getSortKey()).append(" AS SORT_KEY");
+		sql.append("SELECT ").append(topClause).append(" ").append(buildSortKeySelect(provider));
 		sql.append(" FROM ").append(provider.getFromClause());
 		sql.append(provider.getWhereClause() == null ? "" : " WHERE " + provider.getWhereClause());
 		buildGroupByClause(provider, sql);
-		sql.append(" ORDER BY ").append(provider.getSortKeyWithoutAlias());
-		buildAscendingClause(provider, sql);
+		sql.append(" ORDER BY ").append(buildSortClause(provider));
 
 		return sql.toString();
 	}
 
-	private static void buildAscendingClause(AbstractSqlPagingQueryProvider provider, StringBuilder sql) {
-		if (provider.isAscending()) {
-			sql.append(" ASC");
+	/**
+	 * Generates ORDER BY attributes based on the sort keys.
+	 * 
+	 * @param provider
+	 * @return a String that can be appended to an ORDER BY clause.
+	 */
+	public static String buildSortClause(AbstractSqlPagingQueryProvider provider) {
+		StringBuilder builder = new StringBuilder();
+		String prefix = "";
+		
+		for (Map.Entry<String, Boolean> sortKey : provider.getSortKeys().entrySet()) {
+			builder.append(prefix);
+			
+			prefix = ", ";
+			
+			builder.append(sortKey.getKey());
+			
+			if(sortKey.getValue() != null && !sortKey.getValue()) {
+				builder.append(" DESC");
+			}
+			else {
+				builder.append(" ASC");
+			}
 		}
-		else {
-			sql.append(" DESC");
+		
+		return builder.toString();
+	}
+
+	/**
+	 * Appends the where conditions required to query for the subsequent pages.
+	 * 
+	 * @param provider
+	 * @param sql
+	 */
+	public static void buildSortConditions(
+			AbstractSqlPagingQueryProvider provider, StringBuilder sql) {
+		List<Map.Entry<String, Boolean>> keys = new ArrayList<Map.Entry<String,Boolean>>(provider.getSortKeys().entrySet());
+		List<String> clauses = new ArrayList<String>();
+		
+		for(int i = 0; i < keys.size(); i++) {
+			StringBuilder clause = new StringBuilder();
+			
+			String prefix = "";
+			for(int j = 0; j < i; j++) {
+				clause.append(prefix);
+				prefix = " AND ";
+				Entry<String, Boolean> entry = keys.get(j);
+				clause.append(entry.getKey());
+				clause.append(" = ");
+				clause.append(provider.getSortKeyPlaceHolder(entry.getKey()));
+			}
+			
+			if(clause.length() > 0) {
+				clause.append(" AND ");
+			}
+			clause.append(keys.get(i).getKey());
+			
+			if(keys.get(i).getValue() != null && !keys.get(i).getValue()) {
+				clause.append(" < ");
+			}
+			else {
+				clause.append(" > ");
+			}
+
+			clause.append(provider.getSortKeyPlaceHolder(keys.get(i).getKey()));
+			
+			clauses.add(clause.toString());
 		}
+		
+		sql.append("(");
+		String prefix = "";
+		
+		for (String curClause : clauses) {
+			sql.append(prefix);
+			prefix = " OR ";
+			sql.append("(");
+			sql.append(curClause);
+			sql.append(")");
+		}
+		sql.append(")");
+	}
+
+	private static String buildSortKeySelect(AbstractSqlPagingQueryProvider provider) {
+		StringBuilder select = new StringBuilder();
+		
+		String prefix = "";
+		
+		for (Map.Entry<String, Boolean> sortKey : provider.getSortKeys().entrySet()) {
+			select.append(prefix);
+			
+			prefix = ", ";
+			
+			select.append(sortKey.getKey());
+		}
+		
+		return select.toString();
 	}
 
 	private static void buildWhereClause(AbstractSqlPagingQueryProvider provider, boolean remainingPageQuery,
@@ -198,14 +286,8 @@ public class SqlPagingQueryUtils {
 				sql.append(provider.getWhereClause());
 				sql.append(" AND ");
 			}
-			sql.append(provider.getSortKey());
-			if (provider.isAscending()) {
-				sql.append(" > ");
-			}
-			else {
-				sql.append(" < ");
-			}
-			sql.append(provider.getSortKeyPlaceHolder());
+
+			buildSortConditions(provider, sql);
 		}
 		else {
 			sql.append(provider.getWhereClause() == null ? "" : " WHERE " + provider.getWhereClause());

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

@@ -16,6 +16,8 @@
 
 package org.springframework.batch.item.database.support;
 
+import java.util.Map;
+
 import org.springframework.util.StringUtils;
 
 /**
@@ -73,14 +75,9 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 			sql.append(getWhereClause());
 			sql.append(" AND ");
 		}
-		sql.append(getSortKey());
-		if (isAscending()) {
-			sql.append(" > ");
-		}
-		else {
-			sql.append(" < ");
-		}
-		sql.append(getSortKeyPlaceHolder());
+		
+		SqlPagingQueryUtils.buildSortConditions(this, sql);
+
 		sql.append(getGroupClause() == null ? "" : " GROUP BY " + getGroupClause());
 		sql.append(getOverSubstituteClauseEnd());
 		sql.append(") ").append(getSubQueryAlias()).append("WHERE ").append(extractTableAlias()).append(
@@ -98,9 +95,12 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 		}
 
 		StringBuilder sql = new StringBuilder();
-		sql.append("SELECT SORT_KEY FROM ( ");
-		sql.append("SELECT ").append(getSortKey()).append(" AS SORT_KEY, ");
-		sql.append("ROW_NUMBER() OVER (").append(getOverClause());
+		sql.append("SELECT ");
+		buildSortKeySelect(sql);
+		sql.append(" FROM ( ");
+		sql.append("SELECT ");
+		buildSortKeySelect(sql);
+		sql.append(", ROW_NUMBER() OVER (").append(getOverClause());
 		sql.append(") AS ROW_NUMBER");
 		sql.append(getOverSubstituteClauseStart());
 		sql.append(" FROM ").append(getFromClause());
@@ -113,8 +113,21 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 		return sql.toString();
 	}
 
+	private void buildSortKeySelect(StringBuilder sql) {
+		String prefix = "";
+		for (Map.Entry<String, Boolean> sortKey : getSortKeys().entrySet()) {
+			sql.append(prefix);
+			prefix = ", ";
+			sql.append(sortKey.getKey());
+		}
+	}
+
 	protected String getOverClause() {
-		return "ORDER BY " + getSortKeyWithoutAlias() + " " + getAscendingClause();
+		StringBuilder sql = new StringBuilder();
+		
+		sql.append(" ORDER BY ").append(SqlPagingQueryUtils.buildSortClause(this));
+		
+		return sql.toString();
 	}
 
 	protected String getOverSubstituteClauseStart() {
@@ -124,14 +137,4 @@ public class SqlWindowingPagingQueryProvider extends AbstractSqlPagingQueryProvi
 	protected String getOverSubstituteClauseEnd() {
 		return "";
 	}
-
-	private String getAscendingClause() {
-		if (isAscending()) {
-			return "ASC";
-		}
-		else {
-			return "DESC";
-		}
-	}
-
 }