DATAJDBC-101 - Adds support for paging and sorting repositories.

Original pull request: #188.

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.jdbc.core;
 
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.lang.Nullable;
 
 /**
@@ -22,6 +25,7 @@ import org.springframework.lang.Nullable;
  *
  * @author Jens Schauder
  * @author Thomas Lang
+ * @author Milan Milanov
  */
 public interface JdbcAggregateOperations {
 
@@ -128,4 +132,24 @@ public interface JdbcAggregateOperations {
 	 * @return whether the aggregate exists.
 	 */
 	<T> boolean existsById(Object id, Class<T> domainType);
+
+	/**
+	 * Load all aggregates of a given type, sorted.
+	 *
+	 * @param domainType the type of the aggregate roots. Must not be {@code null}.
+	 * @param <T> the type of the aggregate roots. Must not be {@code null}.
+	 * @param sort the sorting information. Must not be {@code null}.
+	 * @return Guaranteed to be not {@code null}.
+	 */
+	<T> Iterable<T> findAll(Class<T> domainType, Sort sort);
+
+	/**
+	 * Load a page of (potentially sorted) aggregates of a given type.
+	 *
+	 * @param domainType the type of the aggregate roots. Must not be {@code null}.
+	 * @param <T> the type of the aggregate roots. Must not be {@code null}.
+	 * @param pageable the pagination information. Must not be {@code null}.
+	 * @return Guaranteed to be not {@code null}.
+	 */
+	<T> Page<T> findAll(Class<T> domainType, Pageable pageable);
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java

@@ -19,9 +19,15 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.PageImpl;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.DataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
 import org.springframework.data.mapping.IdentifierAccessor;
@@ -45,6 +51,7 @@ import org.springframework.util.Assert;
  * @author Mark Paluch
  * @author Thomas Lang
  * @author Christoph Strobl
+ * @author Milan Milanov
  */
 public class JdbcAggregateTemplate implements JdbcAggregateOperations {
 
@@ -223,6 +230,35 @@ public class JdbcAggregateTemplate implements JdbcAggregateOperations {
 		return accessStrategy.existsById(id, domainType);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
+
+		Assert.notNull(domainType, "Domain type must not be null!");
+
+		Iterable<T> all = accessStrategy.findAll(domainType, sort);
+		return triggerAfterLoad(all);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Pageable)
+	 */
+	@Override
+	public <T> Page<T> findAll(Class<T> domainType, Pageable pageable) {
+
+		Assert.notNull(domainType, "Domain type must not be null!");
+
+		Iterable<T> items = triggerAfterLoad(accessStrategy.findAll(domainType, pageable));
+		long totalCount = accessStrategy.count(domainType);
+
+		return new PageImpl<>(StreamSupport.stream(items.spliterator(), false).collect(Collectors.toList()), pageable,
+				totalCount);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class)

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java

@@ -21,6 +21,8 @@ import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.domain.Identifier;
@@ -33,6 +35,7 @@ import org.springframework.data.relational.core.sql.SqlIdentifier;
  * @author Jens Schauder
  * @author Mark Paluch
  * @author Tyler Van Gorder
+ * @author Milan Milanov
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -188,6 +191,24 @@ public class CascadingDataAccessStrategy implements DataAccessStrategy {
 		return collect(das -> das.existsById(id, domainType));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
+		return collect(das -> das.findAll(domainType, sort));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Pageable)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Pageable pageable) {
+		return collect(das -> das.findAll(domainType, pageable));
+	}
+
 	private <T> T collect(Function<DataAccessStrategy, T> function) {
 
 		// Keep <T> as Eclipse fails to compile if <> is used.

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DataAccessStrategy.java

@@ -18,6 +18,8 @@ package org.springframework.data.jdbc.core.convert;
 import java.util.Map;
 
 import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.JdbcAggregateOperations;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -32,6 +34,7 @@ import org.springframework.lang.Nullable;
  *
  * @author Jens Schauder
  * @author Tyler Van Gorder
+ * @author Milan Milanov
  */
 public interface DataAccessStrategy extends RelationResolver {
 
@@ -215,4 +218,24 @@ public interface DataAccessStrategy extends RelationResolver {
 	 * @return {@code true} if a matching row exists, otherwise {@code false}.
 	 */
 	<T> boolean existsById(Object id, Class<T> domainType);
+
+	/**
+	 * Loads all entities of the given type, sorted.
+	 *
+	 * @param domainType the type of entities to load. Must not be {@code null}.
+	 * @param <T> the type of entities to load.
+	 * @param sort the sorting information. Must not be {@code null}.
+	 * @return Guaranteed to be not {@code null}.
+	 */
+	<T> Iterable<T> findAll(Class<T> domainType, Sort sort);
+
+	/**
+	 * Loads all entities of the given type, paged and sorted.
+	 *
+	 * @param domainType the type of entities to load. Must not be {@code null}.
+	 * @param <T> the type of entities to load.
+	 * @param pageable the pagination information. Must not be {@code null}.
+	 * @return Guaranteed to be not {@code null}.
+	 */
+	<T> Iterable<T> findAll(Class<T> domainType, Pageable pageable);
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java

@@ -29,6 +29,8 @@ import org.springframework.dao.DataRetrievalFailureException;
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.support.JdbcUtil;
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
@@ -60,6 +62,7 @@ import org.springframework.util.Assert;
  * @author Christoph Strobl
  * @author Tom Hombergs
  * @author Tyler Van Gorder
+ * @author Milan Milanov
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -374,6 +377,26 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		return result;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Sort)
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
+		return operations.query(sql(domainType).getFindAll(sort), (RowMapper<T>) getEntityRowMapper(domainType));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Pageable)
+	 */
+	@Override
+	@SuppressWarnings("unchecked")
+	public <T> Iterable<T> findAll(Class<T> domainType, Pageable pageable) {
+		return operations.query(sql(domainType).getFindAll(pageable), (RowMapper<T>) getEntityRowMapper(domainType));
+	}
+
 	private <S, T> SqlIdentifierParameterSource getParameterSource(@Nullable S instance,
 			RelationalPersistentEntity<S> persistentEntity, String prefix,
 			Predicate<RelationalPersistentProperty> skipProperty, IdentifierProcessing identifierProcessing) {

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DelegatingDataAccessStrategy.java

@@ -17,6 +17,8 @@ package org.springframework.data.jdbc.core.convert;
 
 import java.util.Map;
 
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.domain.Identifier;
@@ -29,6 +31,7 @@ import org.springframework.util.Assert;
  *
  * @author Jens Schauder
  * @author Tyler Van Gorder
+ * @author Milan Milanov
  * @since 1.1
  */
 public class DelegatingDataAccessStrategy implements DataAccessStrategy {
@@ -187,6 +190,24 @@ public class DelegatingDataAccessStrategy implements DataAccessStrategy {
 		return delegate.existsById(id, domainType);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
+		return delegate.findAll(domainType, sort);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Pageable)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Pageable pageable) {
+		return delegate.findAll(domainType, pageable);
+	}
+
 	/**
 	 * Must be called exactly once before calling any of the other methods.
 	 *

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGenerator.java

@@ -31,10 +31,14 @@ import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.repository.support.SimpleJdbcRepository;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.mapping.PropertyHandler;
 import org.springframework.data.mapping.context.MappingContext;
+import org.springframework.data.relational.core.dialect.Dialect;
+import org.springframework.data.relational.core.dialect.RenderContextFactory;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
@@ -56,6 +60,7 @@ import org.springframework.util.Assert;
  * @author Mark Paluch
  * @author Tom Hombergs
  * @author Tyler Van Gorder
+ * @author Milan Milanov
  */
 class SqlGenerator {
 
@@ -71,6 +76,7 @@ class SqlGenerator {
 	private final IdentifierProcessing identifierProcessing;
 
 	private final SqlContext sqlContext;
+	private final SqlRenderer sqlRenderer;
 	private final Columns columns;
 
 	private final Lazy<String> findOneSql = Lazy.of(this::createFindOneSql);
@@ -93,16 +99,17 @@ class SqlGenerator {
 	 * @param mappingContext must not be {@literal null}.
 	 * @param converter must not be {@literal null}.
 	 * @param entity must not be {@literal null}.
-	 * @param identifierProcessing must not be {@literal null}.
+	 * @param dialect must not be {@literal null}.
 	 */
 	SqlGenerator(RelationalMappingContext mappingContext, JdbcConverter converter, RelationalPersistentEntity<?> entity,
-			IdentifierProcessing identifierProcessing) {
+			Dialect dialect) {
 
 		this.mappingContext = mappingContext;
 		this.converter = converter;
 		this.entity = entity;
-		this.identifierProcessing = identifierProcessing;
-		this.sqlContext = new SqlContext(entity, identifierProcessing);
+		this.identifierProcessing = dialect.getIdentifierProcessing();
+		this.sqlContext = new SqlContext(entity, this.identifierProcessing);
+		this.sqlRenderer = SqlRenderer.create(new RenderContextFactory(dialect).createRenderContext());
 		this.columns = new Columns(entity, mappingContext, converter);
 	}
 
@@ -175,6 +182,26 @@ class SqlGenerator {
 		return findAllSql.get();
 	}
 
+	/**
+	 * Returns a query for selecting all simple properties of an entity, including those for one-to-one relationships,
+	 * sorted by the given parameter.
+	 *
+	 * @return a SQL statement. Guaranteed to be not {@code null}.
+	 */
+	String getFindAll(Sort sort) {
+		return render(selectBuilder(Collections.emptyList(), sort, Pageable.unpaged()).build());
+	}
+
+	/**
+	 * Returns a query for selecting all simple properties of an entity, including those for one-to-one relationships,
+	 * paged and sorted by the given parameter.
+	 *
+	 * @return a SQL statement. Guaranteed to be not {@code null}.
+	 */
+	String getFindAll(Pageable pageable) {
+		return render(selectBuilder(Collections.emptyList(), pageable.getSort(), pageable).build());
+	}
+
 	/**
 	 * Returns a query for selecting all simple properties of an entity, including those for one-to-one relationships.
 	 * Results are limited to those rows referencing some other entity using the column specified by
@@ -392,6 +419,27 @@ class SqlGenerator {
 		return (SelectBuilder.SelectWhere) baseSelect;
 	}
 
+	private SelectBuilder.SelectOrdered selectBuilder(Collection<String> keyColumns, Sort sort, Pageable pageable) {
+		SelectBuilder.SelectWhere baseSelect = this.selectBuilder(keyColumns);
+
+		if (baseSelect instanceof SelectBuilder.SelectFromAndJoin) {
+			if (pageable.isPaged()) {
+				return ((SelectBuilder.SelectFromAndJoin) baseSelect).limitOffset(pageable.getPageSize(), pageable.getOffset())
+						.orderBy(extractOrderByFields(sort));
+			}
+			return ((SelectBuilder.SelectFromAndJoin) baseSelect).orderBy(extractOrderByFields(sort));
+
+		} else if (baseSelect instanceof SelectBuilder.SelectFromAndJoinCondition) {
+			if (pageable.isPaged()) {
+				return ((SelectBuilder.SelectFromAndJoinCondition) baseSelect)
+						.limitOffset(pageable.getPageSize(), pageable.getOffset()).orderBy(extractOrderByFields(sort));
+			}
+			return baseSelect.orderBy(extractOrderByFields(sort));
+		} else {
+			throw new RuntimeException("Unexpected type found!");
+		}
+	}
+
 	/**
 	 * Create a {@link Column} for {@link PersistentPropertyPathExtension}.
 	 *
@@ -588,19 +636,19 @@ class SqlGenerator {
 	}
 
 	private String render(Select select) {
-		return SqlRenderer.create().render(select);
+		return this.sqlRenderer.render(select);
 	}
 
 	private String render(Insert insert) {
-		return SqlRenderer.create().render(insert);
+		return this.sqlRenderer.render(insert);
 	}
 
 	private String render(Update update) {
-		return SqlRenderer.create().render(update);
+		return this.sqlRenderer.render(update);
 	}
 
 	private String render(Delete delete) {
-		return SqlRenderer.create().render(delete);
+		return this.sqlRenderer.render(delete);
 	}
 
 	private Table getTable() {
@@ -615,6 +663,12 @@ class SqlGenerator {
 		return sqlContext.getVersionColumn();
 	}
 
+	private List<OrderByField> extractOrderByFields(Sort sort) {
+		return sort.stream()
+				.map(order -> OrderByField.from(Column.create(order.getProperty(), this.getTable()), order.getDirection()))
+				.collect(Collectors.toList());
+	}
+
 	/**
 	 * Value object representing a {@code JOIN} association.
 	 */

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/SqlGeneratorSource.java

@@ -29,6 +29,7 @@ import org.springframework.util.ConcurrentReferenceHashMap;
  *
  * @author Jens Schauder
  * @author Mark Paluch
+ * @author Milan Milanov
  */
 @RequiredArgsConstructor
 public class SqlGeneratorSource {
@@ -45,9 +46,8 @@ public class SqlGeneratorSource {
 		return dialect;
 	}
 
-
 	SqlGenerator getSqlGenerator(Class<?> domainType) {
 		return CACHE.computeIfAbsent(domainType, t -> new SqlGenerator(context, converter,
-				context.getRequiredPersistentEntity(t), dialect.getIdentifierProcessing()));
+				context.getRequiredPersistentEntity(t), dialect));
 	}
 }

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java

@@ -18,6 +18,7 @@ package org.springframework.data.jdbc.mybatis;
 import static java.util.Arrays.*;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -27,6 +28,8 @@ import org.mybatis.spring.SqlSessionTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.CascadingDataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.DataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.DefaultDataAccessStrategy;
@@ -58,6 +61,7 @@ import org.springframework.util.Assert;
  * @author Oliver Gierke
  * @author Mark Paluch
  * @author Tyler Van Gorder
+ * @author Milan Milanov
  */
 public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 
@@ -337,6 +341,30 @@ public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 		return sqlSession().selectOne(statement, parameter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Sort)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
+		Map<String, Object> additionalContext = new HashMap<>();
+		additionalContext.put("sort", sort);
+		return sqlSession().selectList(namespace(domainType) + ".findAllSorted",
+				new MyBatisContext(null, null, domainType, additionalContext));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.JdbcAggregateOperations#findAll(java.lang.Class, org.springframework.data.domain.Pageable)
+	 */
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Pageable pageable) {
+		Map<String, Object> additionalContext = new HashMap<>();
+		additionalContext.put("pageable", pageable);
+		return sqlSession().selectList(namespace(domainType) + ".findAllPaged",
+				new MyBatisContext(null, null, domainType, additionalContext));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#count(java.lang.Class)

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/repository/support/SimpleJdbcRepository.java

@@ -21,9 +21,13 @@ import lombok.RequiredArgsConstructor;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.JdbcAggregateOperations;
 import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.data.repository.CrudRepository;
+import org.springframework.data.repository.PagingAndSortingRepository;
 import org.springframework.data.util.Streamable;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -32,10 +36,11 @@ import org.springframework.transaction.annotation.Transactional;
  *
  * @author Jens Schauder
  * @author Oliver Gierke
+ * @author Milan Milanov
  */
 @RequiredArgsConstructor
 @Transactional(readOnly = true)
-public class SimpleJdbcRepository<T, ID> implements CrudRepository<T, ID> {
+public class SimpleJdbcRepository<T, ID> implements CrudRepository<T, ID>, PagingAndSortingRepository<T, ID> {
 
 	private final @NonNull JdbcAggregateOperations entityOperations;
 	private final @NonNull PersistentEntity<T, ?> entity;
@@ -144,4 +149,22 @@ public class SimpleJdbcRepository<T, ID> implements CrudRepository<T, ID> {
 	public void deleteAll() {
 		entityOperations.deleteAll(entity.getType());
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.PagingAndSortingRepository#findAll(org.springframework.data.domain.Sort sort)
+	 */
+	@Override
+	public Iterable<T> findAll(Sort sort) {
+		return entityOperations.findAll(entity.getType(), sort);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.repository.PagingAndSortingRepository#findAll(org.springframework.data.domain.Pageable pageable)
+	 */
+	@Override
+	public Page<T> findAll(Pageable pageable) {
+		return entityOperations.findAll(entity.getType(), pageable);
+	}
 }

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/JdbcAggregateTemplateIntegrationTests.java

@@ -50,6 +50,8 @@ import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.ReadOnlyProperty;
 import org.springframework.data.annotation.Version;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.DataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
 import org.springframework.data.jdbc.testing.DatabaseProfileValueSource;
@@ -78,6 +80,7 @@ import org.springframework.transaction.annotation.Transactional;
  * @author Tom Hombergs
  * @author Tyler Van Gorder
  * @author Clemens Hahn
+ * @author Milan Milanov
  */
 @ContextConfiguration
 @Transactional
@@ -89,7 +92,7 @@ public class JdbcAggregateTemplateIntegrationTests {
 
 	@Autowired JdbcAggregateOperations template;
 	@Autowired NamedParameterJdbcOperations jdbcTemplate;
-	LegoSet legoSet = createLegoSet();
+	LegoSet legoSet = createLegoSet("Star Destroyer");
 
 	/**
 	 * creates an instance of {@link NoIdListChain4} with the following properties:
@@ -182,10 +185,10 @@ public class JdbcAggregateTemplateIntegrationTests {
 						.get("current.database.is.not." + dbProfileName)));
 	}
 
-	private static LegoSet createLegoSet() {
+	private static LegoSet createLegoSet(String name) {
 
 		LegoSet entity = new LegoSet();
-		entity.setName("Star Destroyer");
+		entity.setName(name);
 
 		Manual manual = new Manual();
 		manual.setContent("Accelerates to 99% of light speed. Destroys almost everything. See https://what-if.xkcd.com/1/");
@@ -226,6 +229,39 @@ public class JdbcAggregateTemplateIntegrationTests {
 				.contains(tuple(legoSet.getId(), legoSet.getManual().getId(), legoSet.getManual().getContent()));
 	}
 
+	@Test // DATAJDBC-101
+	public void saveAndLoadManyEntitiesWithReferencedEntitySorted() {
+		template.save(createLegoSet("Lava"));
+		template.save(createLegoSet("Star"));
+		template.save(createLegoSet("Frozen"));
+
+		Iterable<LegoSet> reloadedLegoSets = template.findAll(LegoSet.class, Sort.by("name"));
+
+		assertThat(reloadedLegoSets).hasSize(3).extracting("name").isEqualTo(Arrays.asList("Frozen", "Lava", "Star"));
+	}
+
+	@Test // DATAJDBC-101
+	public void saveAndLoadManyEntitiesWithReferencedEntityPaged() {
+		template.save(createLegoSet("Lava"));
+		template.save(createLegoSet("Star"));
+		template.save(createLegoSet("Frozen"));
+
+		Iterable<LegoSet> reloadedLegoSets = template.findAll(LegoSet.class, PageRequest.of(1, 1));
+
+		assertThat(reloadedLegoSets).hasSize(1).extracting("name").isEqualTo(singletonList("Star"));
+	}
+
+	@Test // DATAJDBC-101
+	public void saveAndLoadManyEntitiesWithReferencedEntitySortedAndPaged() {
+		template.save(createLegoSet("Lava"));
+		template.save(createLegoSet("Star"));
+		template.save(createLegoSet("Frozen"));
+
+		Iterable<LegoSet> reloadedLegoSets = template.findAll(LegoSet.class, PageRequest.of(1, 2, Sort.by("name")));
+
+		assertThat(reloadedLegoSets).hasSize(1).extracting("name").isEqualTo(singletonList("Star"));
+	}
+
 	@Test // DATAJDBC-112
 	public void saveAndLoadManyEntitiesByIdWithReferencedEntity() {
 

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/JdbcAggregateTemplateUnitTests.java

@@ -27,11 +27,11 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
-import org.mockito.invocation.InvocationOnMock;
 import org.mockito.junit.MockitoJUnitRunner;
-import org.mockito.stubbing.Answer;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.BasicJdbcConverter;
 import org.springframework.data.jdbc.core.convert.DataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
@@ -53,6 +53,7 @@ import org.springframework.data.relational.core.mapping.event.BeforeSaveCallback
  *
  * @author Christoph Strobl
  * @author Mark Paluch
+ * @author Milan Milanov
  */
 @RunWith(MockitoJUnitRunner.class)
 public class JdbcAggregateTemplateUnitTests {
@@ -145,6 +146,50 @@ public class JdbcAggregateTemplateUnitTests {
 		assertThat(all).containsExactly(alfred2, neumann2);
 	}
 
+	@Test // DATAJDBC-101
+	public void callbackOnLoadSorted() {
+
+		SampleEntity alfred1 = new SampleEntity(23L, "Alfred");
+		SampleEntity alfred2 = new SampleEntity(23L, "Alfred E.");
+
+		SampleEntity neumann1 = new SampleEntity(42L, "Neumann");
+		SampleEntity neumann2 = new SampleEntity(42L, "Alfred E. Neumann");
+
+		when(dataAccessStrategy.findAll(SampleEntity.class, Sort.by("name"))).thenReturn(asList(alfred1, neumann1));
+
+		when(callbacks.callback(any(Class.class), eq(alfred1), any())).thenReturn(alfred2);
+		when(callbacks.callback(any(Class.class), eq(neumann1), any())).thenReturn(neumann2);
+
+		Iterable<SampleEntity> all = template.findAll(SampleEntity.class, Sort.by("name"));
+
+		verify(callbacks).callback(AfterLoadCallback.class, alfred1);
+		verify(callbacks).callback(AfterLoadCallback.class, neumann1);
+
+		assertThat(all).containsExactly(alfred2, neumann2);
+	}
+
+	@Test // DATAJDBC-101
+	public void callbackOnLoadPaged() {
+
+		SampleEntity alfred1 = new SampleEntity(23L, "Alfred");
+		SampleEntity alfred2 = new SampleEntity(23L, "Alfred E.");
+
+		SampleEntity neumann1 = new SampleEntity(42L, "Neumann");
+		SampleEntity neumann2 = new SampleEntity(42L, "Alfred E. Neumann");
+
+		when(dataAccessStrategy.findAll(SampleEntity.class, PageRequest.of(0, 20))).thenReturn(asList(alfred1, neumann1));
+
+		when(callbacks.callback(any(Class.class), eq(alfred1), any())).thenReturn(alfred2);
+		when(callbacks.callback(any(Class.class), eq(neumann1), any())).thenReturn(neumann2);
+
+		Iterable<SampleEntity> all = template.findAll(SampleEntity.class, PageRequest.of(0, 20));
+
+		verify(callbacks).callback(AfterLoadCallback.class, alfred1);
+		verify(callbacks).callback(AfterLoadCallback.class, neumann1);
+
+		assertThat(all).containsExactly(alfred2, neumann2);
+	}
+
 	@Data
 	@AllArgsConstructor
 	private static class SampleEntity {

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorContextBasedNamingStrategyUnitTests.java

@@ -27,14 +27,12 @@ import org.junit.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
 import org.springframework.data.jdbc.core.mapping.PersistentPropertyPathTestUtils;
+import org.springframework.data.jdbc.testing.NonQuotingDialect;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.NamingStrategy;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
-import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
-import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
 
 /**
  * Unit tests to verify a contextual {@link NamingStrategy} implementation that customizes using a user-centric
@@ -221,8 +219,7 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 		});
 		RelationalPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(DummyEntity.class);
 
-		return new SqlGenerator(context, converter, persistentEntity,
-				IdentifierProcessing.create(new Quoting(""), LetterCasing.AS_IS));
+		return new SqlGenerator(context, converter, persistentEntity, NonQuotingDialect.INSTANCE);
 	}
 
 	@SuppressWarnings("unused")

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorEmbeddedUnitTests.java

@@ -25,6 +25,7 @@ import org.junit.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.PropertyPathTestingUtils;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
+import org.springframework.data.jdbc.testing.NonQuotingDialect;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.Embedded;
 import org.springframework.data.relational.core.mapping.Embedded.OnEmpty;
@@ -32,9 +33,6 @@ import org.springframework.data.relational.core.mapping.PersistentPropertyPathEx
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.sql.Aliased;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
-import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
-import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
 
 /**
  * Unit tests for the {@link SqlGenerator} in a context of the {@link Embedded} annotation.
@@ -56,8 +54,7 @@ public class SqlGeneratorEmbeddedUnitTests {
 
 	SqlGenerator createSqlGenerator(Class<?> type) {
 		RelationalPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(type);
-		return new SqlGenerator(context, converter, persistentEntity,
-				IdentifierProcessing.create(new Quoting(""), LetterCasing.AS_IS));
+		return new SqlGenerator(context, converter, persistentEntity, NonQuotingDialect.INSTANCE);
 	}
 
 	@Test // DATAJDBC-111

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorFixedNamingStrategyUnitTests.java

@@ -23,12 +23,12 @@ import org.junit.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
 import org.springframework.data.jdbc.core.mapping.PersistentPropertyPathTestUtils;
+import org.springframework.data.jdbc.testing.AnsiDialect;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.NamingStrategy;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
 
 /**
  * Unit tests the {@link SqlGenerator} with a fixed {@link NamingStrategy} implementation containing a hard wired
@@ -199,7 +199,7 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 			throw new UnsupportedOperationException();
 		});
 		RelationalPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(DummyEntity.class);
-		return new SqlGenerator(context, converter, persistentEntity, IdentifierProcessing.ANSI);
+		return new SqlGenerator(context, converter, persistentEntity, AnsiDialect.INSTANCE);
 	}
 
 	@SuppressWarnings("unused")

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java

@@ -29,11 +29,17 @@ import org.junit.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.ReadOnlyProperty;
 import org.springframework.data.annotation.Version;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.PropertyPathTestingUtils;
 import org.springframework.data.jdbc.core.mapping.AggregateReference;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
 import org.springframework.data.jdbc.core.mapping.PersistentPropertyPathTestUtils;
+import org.springframework.data.jdbc.testing.AnsiDialect;
+import org.springframework.data.jdbc.testing.NonQuotingDialect;
 import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.NamingStrategy;
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
@@ -43,9 +49,6 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
 import org.springframework.data.relational.core.sql.Aliased;
 import org.springframework.data.relational.core.sql.Table;
 import org.springframework.data.relational.domain.Identifier;
-import org.springframework.data.relational.core.sql.IdentifierProcessing;
-import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
-import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
 
 /**
  * Unit tests for the {@link SqlGenerator}.
@@ -56,6 +59,7 @@ import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting
  * @author Bastian Wilhelm
  * @author Mark Paluch
  * @author Tom Hombergs
+ * @author Milan Milanov
  */
 public class SqlGeneratorUnitTests {
 
@@ -75,14 +79,14 @@ public class SqlGeneratorUnitTests {
 
 	SqlGenerator createSqlGenerator(Class<?> type) {
 
-		return createSqlGenerator(type, IdentifierProcessing.create(new Quoting(""), LetterCasing.AS_IS));
+		return createSqlGenerator(type, NonQuotingDialect.INSTANCE);
 	}
 
-	SqlGenerator createSqlGenerator(Class<?> type, IdentifierProcessing identifierProcessing) {
+	SqlGenerator createSqlGenerator(Class<?> type, Dialect dialect) {
 
 		RelationalPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(type);
 
-		return new SqlGenerator(context, converter, persistentEntity, identifierProcessing);
+		return new SqlGenerator(context, converter, persistentEntity, dialect);
 	}
 
 	@Test // DATAJDBC-112
@@ -163,6 +167,103 @@ public class SqlGeneratorUnitTests {
 		assertThat(sql).isEqualTo("DELETE FROM element WHERE element.dummy_entity = :rootId");
 	}
 
+	@Test // DATAJDBC-101
+	public void findAllSortedByUnsorted() {
+
+		String sql = sqlGenerator.getFindAll(Sort.unsorted());
+
+		assertThat(sql).doesNotContain("ORDER BY");
+	}
+
+	@Test // DATAJDBC-101
+	public void findAllSortedBySingleField() {
+
+		String sql = sqlGenerator.getFindAll(Sort.by("x_name"));
+
+		assertThat(sql).contains("SELECT", //
+				"dummy_entity.id1 AS id1", //
+				"dummy_entity.x_name AS x_name", //
+				"dummy_entity.x_other AS x_other", //
+				"ref.x_l1id AS ref_x_l1id", //
+				"ref.x_content AS ref_x_content", //
+				"ref_further.x_l2id AS ref_further_x_l2id", //
+				"ref_further.x_something AS ref_further_x_something", //
+				"FROM dummy_entity ", //
+				"LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1", //
+				"LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id", //
+				"ORDER BY x_name ASC");
+	}
+
+	@Test // DATAJDBC-101
+	public void findAllSortedByMultipleFields() {
+
+		String sql = sqlGenerator.getFindAll(
+				Sort.by(new Sort.Order(Sort.Direction.DESC, "x_name"), new Sort.Order(Sort.Direction.ASC, "x_other")));
+
+		assertThat(sql).contains("SELECT", //
+				"dummy_entity.id1 AS id1", //
+				"dummy_entity.x_name AS x_name", //
+				"dummy_entity.x_other AS x_other", //
+				"ref.x_l1id AS ref_x_l1id", //
+				"ref.x_content AS ref_x_content", //
+				"ref_further.x_l2id AS ref_further_x_l2id", //
+				"ref_further.x_something AS ref_further_x_something", //
+				"FROM dummy_entity ", //
+				"LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1", //
+				"LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id", //
+				"ORDER BY x_name DESC", //
+				"x_other ASC");
+	}
+
+	@Test // DATAJDBC-101
+	public void findAllPagedByUnpaged() {
+
+		String sql = sqlGenerator.getFindAll(Pageable.unpaged());
+
+		assertThat(sql).doesNotContain("ORDER BY").doesNotContain("FETCH FIRST").doesNotContain("OFFSET");
+	}
+
+	@Test // DATAJDBC-101
+	public void findAllPaged() {
+
+		String sql = sqlGenerator.getFindAll(PageRequest.of(2, 20));
+
+		assertThat(sql).contains("SELECT", //
+				"dummy_entity.id1 AS id1", //
+				"dummy_entity.x_name AS x_name", //
+				"dummy_entity.x_other AS x_other", //
+				"ref.x_l1id AS ref_x_l1id", //
+				"ref.x_content AS ref_x_content", //
+				"ref_further.x_l2id AS ref_further_x_l2id", //
+				"ref_further.x_something AS ref_further_x_something", //
+				"FROM dummy_entity ", //
+				"LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1", //
+				"LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id", //
+				"OFFSET 40 ROWS", //
+				"FETCH FIRST 20 ROWS ONLY");
+	}
+
+	@Test // DATAJDBC-101
+	public void findAllPagedAndSorted() {
+
+		String sql = sqlGenerator.getFindAll(PageRequest.of(3, 10, Sort.by("x_name")));
+
+		assertThat(sql).contains("SELECT", //
+				"dummy_entity.id1 AS id1", //
+				"dummy_entity.x_name AS x_name", //
+				"dummy_entity.x_other AS x_other", //
+				"ref.x_l1id AS ref_x_l1id", //
+				"ref.x_content AS ref_x_content", //
+				"ref_further.x_l2id AS ref_further_x_l2id", //
+				"ref_further.x_something AS ref_further_x_something", //
+				"FROM dummy_entity ", //
+				"LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1", //
+				"LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id", //
+				"ORDER BY x_name ASC", //
+				"OFFSET 30 ROWS", //
+				"FETCH FIRST 10 ROWS ONLY");
+	}
+
 	@Test // DATAJDBC-131, DATAJDBC-111
 	public void findAllByProperty() {
 
@@ -248,7 +349,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-219
 	public void updateWithVersion() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(VersionedEntity.class, IdentifierProcessing.ANSI);
+		SqlGenerator sqlGenerator = createSqlGenerator(VersionedEntity.class, AnsiDialect.INSTANCE);
 
 		assertThat(sqlGenerator.getUpdateWithVersion()).containsSequence( //
 				"UPDATE", //
@@ -273,7 +374,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-334
 	public void getInsertForQuotedColumnName() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(EntityWithQuotedColumnName.class, IdentifierProcessing.ANSI);
+		SqlGenerator sqlGenerator = createSqlGenerator(EntityWithQuotedColumnName.class, AnsiDialect.INSTANCE);
 
 		String insert = sqlGenerator.getInsert(emptySet());
 
@@ -284,7 +385,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-266
 	public void joinForOneToOneWithoutIdIncludesTheBackReferenceOfTheOuterJoin() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(ParentOfNoIdChild.class, IdentifierProcessing.ANSI);
+		SqlGenerator sqlGenerator = createSqlGenerator(ParentOfNoIdChild.class, AnsiDialect.INSTANCE);
 
 		String findAll = sqlGenerator.getFindAll();
 
@@ -295,7 +396,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-262
 	public void update() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntity.class, IdentifierProcessing.ANSI);
+		SqlGenerator sqlGenerator = createSqlGenerator(DummyEntity.class, AnsiDialect.INSTANCE);
 
 		assertThat(sqlGenerator.getUpdate()).containsSequence( //
 				"UPDATE", //
@@ -308,7 +409,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-324
 	public void readOnlyPropertyExcludedFromQuery_when_generateUpdateSql() {
 
-		final SqlGenerator sqlGenerator = createSqlGenerator(EntityWithReadOnlyProperty.class, IdentifierProcessing.ANSI);
+		final SqlGenerator sqlGenerator = createSqlGenerator(EntityWithReadOnlyProperty.class, AnsiDialect.INSTANCE);
 
 		assertThat(sqlGenerator.getUpdate()).isEqualToIgnoringCase( //
 				"UPDATE \"ENTITY_WITH_READ_ONLY_PROPERTY\" " //
@@ -320,7 +421,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-334
 	public void getUpdateForQuotedColumnName() {
 
-		SqlGenerator sqlGenerator = createSqlGenerator(EntityWithQuotedColumnName.class, IdentifierProcessing.ANSI);
+		SqlGenerator sqlGenerator = createSqlGenerator(EntityWithQuotedColumnName.class, AnsiDialect.INSTANCE);
 
 		String update = sqlGenerator.getUpdate();
 
@@ -332,7 +433,7 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-324
 	public void readOnlyPropertyExcludedFromQuery_when_generateInsertSql() {
 
-		final SqlGenerator sqlGenerator = createSqlGenerator(EntityWithReadOnlyProperty.class, IdentifierProcessing.ANSI);
+		final SqlGenerator sqlGenerator = createSqlGenerator(EntityWithReadOnlyProperty.class, AnsiDialect.INSTANCE);
 
 		assertThat(sqlGenerator.getInsert(emptySet())).isEqualToIgnoringCase( //
 				"INSERT INTO \"ENTITY_WITH_READ_ONLY_PROPERTY\" (\"X_NAME\") " //
@@ -514,7 +615,7 @@ public class SqlGeneratorUnitTests {
 	}
 
 	private SqlGenerator.Join generateJoin(String path, Class<?> type) {
-		return createSqlGenerator(type, IdentifierProcessing.ANSI)
+		return createSqlGenerator(type, AnsiDialect.INSTANCE)
 				.getJoin(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
 	}
 
@@ -559,7 +660,7 @@ public class SqlGeneratorUnitTests {
 
 	private org.springframework.data.relational.core.sql.Column generatedColumn(String path, Class<?> type) {
 
-		return createSqlGenerator(type, IdentifierProcessing.ANSI)
+		return createSqlGenerator(type, AnsiDialect.INSTANCE)
 				.getColumn(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
 	}
 

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategyUnitTests.java

@@ -30,6 +30,8 @@ import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.PropertyPathTestingUtils;
 import org.springframework.data.jdbc.core.convert.BasicJdbcConverter;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
@@ -400,6 +402,50 @@ public class MyBatisDataAccessStrategyUnitTests {
 				);
 	}
 
+	@Test // DATAJDBC-101
+	public void findAllSorted() {
+
+		accessStrategy.findAll(String.class, Sort.by("length"));
+
+		verify(session).selectList(eq("java.lang.StringMapper.findAllSorted"), captor.capture());
+
+		assertThat(captor.getValue()) //
+				.isNotNull() //
+				.extracting( //
+						MyBatisContext::getInstance, //
+						MyBatisContext::getId, //
+						MyBatisContext::getDomainType, //
+						c -> c.get("sort") //
+				).containsExactly( //
+						null, //
+						null, //
+						String.class, //
+						Sort.by("length") //
+				);
+	}
+
+	@Test // DATAJDBC-101
+	public void findAllPaged() {
+
+		accessStrategy.findAll(String.class, PageRequest.of(0, 20));
+
+		verify(session).selectList(eq("java.lang.StringMapper.findAllPaged"), captor.capture());
+
+		assertThat(captor.getValue()) //
+				.isNotNull() //
+				.extracting( //
+						MyBatisContext::getInstance, //
+						MyBatisContext::getId, //
+						MyBatisContext::getDomainType, //
+						c -> c.get("pageable") //
+				).containsExactly( //
+						null, //
+						null, //
+						String.class, //
+						PageRequest.of(0, 20) //
+				);
+	}
+
 	@SuppressWarnings("unused")
 	private static class DummyEntity {
 		ChildOne one;

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/SimpleJdbcRepositoryEventsUnitTests.java

@@ -35,6 +35,9 @@ import org.junit.Test;
 import org.mockito.stubbing.Answer;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.PageRequest;
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.BasicJdbcConverter;
 import org.springframework.data.jdbc.core.convert.DefaultDataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.DefaultJdbcTypeFactory;
@@ -54,7 +57,7 @@ import org.springframework.data.relational.core.mapping.event.BeforeDeleteEvent;
 import org.springframework.data.relational.core.mapping.event.BeforeSaveEvent;
 import org.springframework.data.relational.core.mapping.event.Identifier;
 import org.springframework.data.relational.core.mapping.event.RelationalEvent;
-import org.springframework.data.repository.CrudRepository;
+import org.springframework.data.repository.PagingAndSortingRepository;
 import org.springframework.jdbc.core.JdbcOperations;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.core.namedparam.SqlParameterSource;
@@ -67,6 +70,7 @@ import org.springframework.jdbc.support.KeyHolder;
  * @author Mark Paluch
  * @author Oliver Gierke
  * @author Myeonghyeon Lee
+ * @author Milan Milanov
  */
 public class SimpleJdbcRepositoryEventsUnitTests {
 
@@ -216,6 +220,45 @@ public class SimpleJdbcRepositoryEventsUnitTests {
 				);
 	}
 
+	@Test // DATAJDBC-101
+	@SuppressWarnings("rawtypes")
+	public void publishesEventsOnFindAllSorted() {
+
+		DummyEntity entity1 = new DummyEntity(42L);
+		DummyEntity entity2 = new DummyEntity(23L);
+
+		doReturn(asList(entity1, entity2)).when(dataAccessStrategy).findAll(any(), any(Sort.class));
+
+		repository.findAll(Sort.by("field"));
+
+		assertThat(publisher.events) //
+				.extracting(e -> (Class) e.getClass()) //
+				.containsExactly( //
+						AfterLoadEvent.class, //
+						AfterLoadEvent.class //
+				);
+	}
+
+	@Test // DATAJDBC-101
+	@SuppressWarnings("rawtypes")
+	public void publishesEventsOnFindAllPaged() {
+
+		DummyEntity entity1 = new DummyEntity(42L);
+		DummyEntity entity2 = new DummyEntity(23L);
+
+		doReturn(asList(entity1, entity2)).when(dataAccessStrategy).findAll(any(), any(Pageable.class));
+		doReturn(2L).when(dataAccessStrategy).count(any());
+
+		repository.findAll(PageRequest.of(0, 20));
+
+		assertThat(publisher.events) //
+				.extracting(e -> (Class) e.getClass()) //
+				.containsExactly( //
+						AfterLoadEvent.class, //
+						AfterLoadEvent.class //
+				);
+	}
+
 	private static NamedParameterJdbcOperations createIdGeneratingOperations() {
 
 		Answer<Integer> setIdInKeyHolder = invocation -> {
@@ -235,7 +278,7 @@ public class SimpleJdbcRepositoryEventsUnitTests {
 		return operations;
 	}
 
-	interface DummyEntityRepository extends CrudRepository<DummyEntity, Long> {}
+	interface DummyEntityRepository extends PagingAndSortingRepository<DummyEntity, Long> {}
 
 	@Value
 	@With

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/AnsiDialect.java

@@ -0,0 +1,130 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.jdbc.testing;
+
+import lombok.RequiredArgsConstructor;
+
+import org.springframework.data.relational.core.dialect.AbstractDialect;
+import org.springframework.data.relational.core.dialect.ArrayColumns;
+import org.springframework.data.relational.core.dialect.LimitClause;
+import org.springframework.data.relational.core.sql.IdentifierProcessing;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * An SQL dialect for the ANSI SQL standard.
+ *
+ * @author Milan Milanov
+ * @since 2.0
+ */
+public class AnsiDialect extends AbstractDialect {
+
+	/**
+	 * Singleton instance.
+	 */
+	public static final AnsiDialect INSTANCE = new AnsiDialect();
+
+	protected AnsiDialect() {}
+
+	private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LimitClause#getLimit(long)
+		 */
+		@Override
+		public String getLimit(long limit) {
+			return String.format("FETCH FIRST %d ROWS ONLY", limit);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LimitClause#getOffset(long)
+		 */
+		@Override
+		public String getOffset(long offset) {
+			return String.format("OFFSET %d ROWS", offset);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LimitClause#getClause(long, long)
+		 */
+		@Override
+		public String getLimitOffset(long limit, long offset) {
+			return String.format("OFFSET %d ROWS FETCH FIRST %d ROWS ONLY", offset, limit);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LimitClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_ORDER_BY;
+		}
+	};
+
+	private final AnsiArrayColumns ARRAY_COLUMNS = new AnsiArrayColumns();
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#limit()
+	 */
+	@Override
+	public LimitClause limit() {
+		return LIMIT_CLAUSE;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#getArraySupport()
+	 */
+	@Override
+	public ArrayColumns getArraySupport() {
+		return ARRAY_COLUMNS;
+	}
+
+	@RequiredArgsConstructor
+	static class AnsiArrayColumns implements ArrayColumns {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.ArrayColumns#isSupported()
+		 */
+		@Override
+		public boolean isSupported() {
+			return true;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.ArrayColumns#getArrayType(java.lang.Class)
+		 */
+		@Override
+		public Class<?> getArrayType(Class<?> userType) {
+
+			Assert.notNull(userType, "Array component type must not be null");
+
+			return ClassUtils.resolvePrimitiveIfNecessary(userType);
+		}
+	}
+
+	@Override
+	public IdentifierProcessing getIdentifierProcessing() {
+		return IdentifierProcessing.ANSI;
+	}
+}

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/NonQuotingDialect.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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.data.jdbc.testing;
+
+
+import org.springframework.data.relational.core.sql.IdentifierProcessing;
+
+/**
+ * The ANSI standard dialect, but without quoting the identifiers.
+ *
+ * @author Milan Milanov
+ * @since 2.0
+ */
+public class NonQuotingDialect extends AnsiDialect {
+
+	public static final NonQuotingDialect INSTANCE = new NonQuotingDialect();
+
+	@Override
+	public IdentifierProcessing getIdentifierProcessing() {
+		return IdentifierProcessing.NONE;
+	}
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/OrderByField.java

@@ -25,6 +25,7 @@ import org.springframework.util.Assert;
  * Represents a field in the {@code ORDER BY} clause.
  *
  * @author Mark Paluch
+ * @author Milan Milanov
  * @since 1.1
  */
 public class OrderByField extends AbstractSegment {
@@ -54,6 +55,17 @@ public class OrderByField extends AbstractSegment {
 		return new OrderByField(column, null, NullHandling.NATIVE);
 	}
 
+	/**
+	 * Creates a new {@link OrderByField} from a {@link Column} applying a given ordering.
+	 *
+	 * @param column must not be {@literal null}.
+	 * @param direction order direction
+	 * @return the {@link OrderByField}.
+	 */
+	public static OrderByField from(Column column, Direction direction) {
+		return new OrderByField(column, direction, NullHandling.NATIVE);
+	}
+
 	/**
 	 * Creates a new {@link OrderByField} from a the current one using ascending sorting.
 	 *

src/main/asciidoc/jdbc.adoc

@@ -585,6 +585,14 @@ Note that the type used for prefixing the statement name is the name of the aggr
 
 `getDomainType`: The type of the entity to load.
 
+| `findAllSorted` | Select all aggregate roots, sorted | `findAll(Sort)`.|
+
+`getSort`: The sorting specification.
+
+| `findAllPaged` | Select a page of aggregate roots, optionally sorted | `findAll(Page)`.|
+
+`getPageable`: The paging specification.
+
 | `count` | Count the number of aggregate root of the type used as prefix | `count` |
 
 `getDomainType`: The type of aggregate roots to count.

src/main/asciidoc/new-features.adoc

@@ -7,6 +7,7 @@ This section covers the significant changes for each version.
 == What's New in Spring Data JDBC 2.0
 
 * Optimistic Locking support.
+* Support for `PagingAndSortingRepository`
 
 [[new-features.1-1-0]]
 == What's New in Spring Data JDBC 1.1