Query#scroll support

+ HHH-14308: Add generic type parameter to ScrollableResults

hibernate-core/src/main/java/org/hibernate/ScrollableResults.java

@@ -59,6 +59,13 @@ public interface ScrollableResults<R> extends AutoCloseable, Closeable {
 	 */
 	boolean scroll(int positions);
 
+	/**
+	 * Moves the result cursor to the specified position.
+	 *
+	 * @return {@code true} if there is a result at the new location
+	 */
+	boolean position(int position);
+
 	/**
 	 * Go to the last result.
 	 *

hibernate-core/src/main/java/org/hibernate/internal/AbstractScrollableResults.java

@@ -16,7 +16,8 @@ import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
 import org.hibernate.sql.results.spi.RowReader;
 
 /**
- * Base implementation of the ScrollableResults interface.
+ * Base implementation of the ScrollableResults interface intended for sharing between
+ * {@link ScrollableResultsImpl} and {@link FetchingScrollableResultsImpl}
  *
  * @author Steve Ebersole
  */
@@ -91,7 +92,9 @@ public abstract class AbstractScrollableResults<R> implements ScrollableResultsI
 			return;
 		}
 
-		getJdbcValues().finishUp( persistenceContext );
+		rowReader.finishUp( jdbcValuesSourceProcessingState );
+		jdbcValues.finishUp( persistenceContext );
+
 		getPersistenceContext().getJdbcCoordinator().afterStatementExecution();
 
 		this.closed = true;

hibernate-core/src/main/java/org/hibernate/internal/EmptyScrollableResults.java

@@ -45,6 +45,11 @@ public class EmptyScrollableResults implements ScrollableResultsImplementor {
 		return false;
 	}
 
+	@Override
+	public boolean position(int position) {
+		return false;
+	}
+
 	@Override
 	public boolean last() {
 		return true;

hibernate-core/src/main/java/org/hibernate/internal/FetchingScrollableResultsImpl.java

@@ -155,6 +155,11 @@ public class FetchingScrollableResultsImpl<R> extends AbstractScrollableResults<
 		return more;
 	}
 
+	@Override
+	public boolean position(int position) {
+		throw new NotYetImplementedFor6Exception( getClass() );
+	}
+
 	@Override
 	public boolean last() {
 		throw new NotYetImplementedFor6Exception( getClass() );

hibernate-core/src/main/java/org/hibernate/internal/ScrollableResultsImpl.java

@@ -6,14 +6,11 @@
  */
 package org.hibernate.internal;
 
-import java.sql.SQLException;
-
 import org.hibernate.HibernateException;
-import org.hibernate.JDBCException;
 import org.hibernate.NotYetImplementedFor6Exception;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
-import org.hibernate.sql.results.jdbc.internal.JdbcValuesSourceProcessingStateStandardImpl;
 import org.hibernate.sql.results.internal.RowProcessingStateStandardImpl;
+import org.hibernate.sql.results.jdbc.internal.JdbcValuesSourceProcessingStateStandardImpl;
 import org.hibernate.sql.results.jdbc.spi.JdbcValues;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
 import org.hibernate.sql.results.spi.RowReader;
@@ -49,44 +46,45 @@ public class ScrollableResultsImpl<R> extends AbstractScrollableResults<R> {
 	}
 
 	@Override
-	public boolean scroll(int i) {
+	public boolean next() {
-		throw new NotYetImplementedFor6Exception();
+		final boolean result = getRowProcessingState().next();
-
+		prepareCurrentRow( result );
-		// todo (6.0) : need these scrollable ResultSet "re-positioning"-style methods on the JdbcValues stuff
+		return result;
-
-//		try {
-//			final boolean result = getResultSet().relative( i );
-//			prepareCurrentRow( result );
-//			return result;
-//		}
-//		catch (SQLException sqle) {
-//			throw convert( sqle, "could not advance using scroll()" );
-//		}
 	}
 
-	protected JDBCException convert(SQLException sqle, String message) {
+	@Override
-		return getPersistenceContext().getJdbcServices().getSqlExceptionHelper().convert( sqle, message );
+	public boolean previous() {
+		final boolean result = getRowProcessingState().previous();
+		prepareCurrentRow( result );
+		return result;
+	}
+
+	@Override
+	public boolean scroll(int i) {
+		final boolean hasResult = getRowProcessingState().scroll( i );
+		prepareCurrentRow( hasResult );
+		return hasResult;
+	}
+
+	@Override
+	public boolean position(int position) {
+		final boolean hasResult = getRowProcessingState().position( position );
+		prepareCurrentRow( hasResult );
+		return hasResult;
 	}
 
 	@Override
 	public boolean first() {
-		throw new NotYetImplementedFor6Exception();
+		final boolean hasResult = getRowProcessingState().first();
-
+		prepareCurrentRow( hasResult );
-		// todo (6.0) : need these scrollable ResultSet "re-positioning"-style methods on the JdbcValues stuff
+		return hasResult;
-
-//		try {
-//			final boolean result = getResultSet().first();
-//			prepareCurrentRow( result );
-//			return result;
-//		}
-//		catch (SQLException sqle) {
-//			throw convert( sqle, "could not advance using first()" );
-//		}
 	}
 
 	@Override
 	public boolean last() {
-		throw new NotYetImplementedFor6Exception();
+		final boolean hasResult = getRowProcessingState().last();
+		prepareCurrentRow( hasResult );
+		return hasResult;
 
 		// todo (6.0) : need these scrollable ResultSet "re-positioning"-style methods on the JdbcValues stuff
 
@@ -100,34 +98,6 @@ public class ScrollableResultsImpl<R> extends AbstractScrollableResults<R> {
 //		}
 	}
 
-	@Override
-	public boolean next() {
-		try {
-			final boolean result = getJdbcValues().next( getRowProcessingState() );
-			prepareCurrentRow( result );
-			return result;
-		}
-		catch (SQLException sqle) {
-			throw convert( sqle, "could not advance using next()" );
-		}
-	}
-
-	@Override
-	public boolean previous() {
-		throw new NotYetImplementedFor6Exception();
-
-		// todo (6.0) : need these scrollable ResultSet "re-positioning"-style methods on the JdbcValues stuff
-
-//		try {
-//			final boolean result = getResultSet().previous();
-//			prepareCurrentRow( result );
-//			return result;
-//		}
-//		catch (SQLException sqle) {
-//			throw convert( sqle, "could not advance using previous()" );
-//		}
-	}
-
 	@Override
 	public void afterLast() {
 		throw new NotYetImplementedFor6Exception();
@@ -186,36 +156,12 @@ public class ScrollableResultsImpl<R> extends AbstractScrollableResults<R> {
 
 	@Override
 	public int getRowNumber() throws HibernateException {
-		throw new NotYetImplementedFor6Exception();
+		return getRowProcessingState().getPosition();
-
-		// todo (6.0) : need these scrollable ResultSet "re-positioning"-style methods on the JdbcValues stuff
-
-//		try {
-//			return getResultSet().getRow() - 1;
-//		}
-//		catch (SQLException sqle) {
-//			throw convert( sqle, "exception calling getRow()" );
-//		}
 	}
 
 	@Override
 	public boolean setRowNumber(int rowNumber) throws HibernateException {
-		throw new NotYetImplementedFor6Exception();
+		return position( rowNumber );
-
-		// todo (6.0) : need these scrollable ResultSet "re-positioning"-style methods on the JdbcValues stuff
-
-//		if ( rowNumber >= 0 ) {
-//			rowNumber++;
-//		}
-//
-//		try {
-//			final boolean result = getResultSet().absolute( rowNumber );
-//			prepareCurrentRow( result );
-//			return result;
-//		}
-//		catch (SQLException sqle) {
-//			throw convert( sqle, "could not advance using absolute()" );
-//		}
 	}
 
 	private void prepareCurrentRow(boolean underlyingScrollSuccessful) {
@@ -224,15 +170,10 @@ public class ScrollableResultsImpl<R> extends AbstractScrollableResults<R> {
 			return;
 		}
 
-		try {
+		currentRow = getRowReader().readRow(
-			currentRow = getRowReader().readRow(
+				getRowProcessingState(),
-					getRowProcessingState(),
+				getProcessingOptions()
-					getProcessingOptions()
+		);
-			);
-		}
-		catch (SQLException e) {
-			throw convert( e, "Unable to read row as part of ScrollableResult handling" );
-		}
 
 		afterScrollOperation();
 	}

hibernate-core/src/main/java/org/hibernate/jdbc/ReturningWork.java

@@ -14,6 +14,7 @@ import java.sql.SQLException;
  *
  * @author Steve Ebersole
  */
+@FunctionalInterface
 public interface ReturningWork<T> {
 	/**
 	 * Execute the discrete work encapsulated by this work instance using the supplied connection.
@@ -25,5 +26,5 @@ public interface ReturningWork<T> {
 	 * @throws SQLException Thrown during execution of the underlying JDBC interaction.
 	 * @throws org.hibernate.HibernateException Generally indicates a wrapped SQLException.
 	 */
-	public T execute(Connection connection) throws SQLException;
+	T execute(Connection connection) throws SQLException;
 }

hibernate-core/src/main/java/org/hibernate/jdbc/Work.java

@@ -14,6 +14,7 @@ import java.sql.SQLException;
  *
  * @author Steve Ebersole
  */
+@FunctionalInterface
 public interface Work {
 	/**
 	 * Execute the discrete work encapsulated by this work instance using the supplied connection.

hibernate-core/src/main/java/org/hibernate/procedure/internal/ProcedureCallImpl.java

@@ -747,7 +747,7 @@ public class ProcedureCallImpl<R>
 	}
 
 	@Override
-	protected ScrollableResultsImplementor doScroll(ScrollMode scrollMode) {
+	public ScrollableResultsImplementor scroll(ScrollMode scrollMode) {
 		throw new UnsupportedOperationException( "Query#scroll is not valid for ProcedureCall/StoredProcedureQuery" );
 	}
 

hibernate-core/src/main/java/org/hibernate/query/Query.java

@@ -32,6 +32,7 @@ import org.hibernate.NonUniqueResultException;
 import org.hibernate.ScrollMode;
 import org.hibernate.ScrollableResults;
 import org.hibernate.SharedSessionContract;
+import org.hibernate.dialect.Dialect;
 import org.hibernate.graph.GraphSemantic;
 import org.hibernate.graph.RootGraph;
 import org.hibernate.metamodel.model.domain.AllowableParameterType;
@@ -99,29 +100,23 @@ public interface Query<R> extends TypedQuery<R>, CommonQueryContract {
 	}
 
 	/**
-	 * Return the query results as <tt>ScrollableResults</tt>. The
+	 * Returns scrollable access to the query results.
-	 * scrollability of the returned results depends upon JDBC driver
-	 * support for scrollable <tt>ResultSet</tt>s.<br>
 	 *
-	 * @see ScrollableResults
+	 * This form calls {@link #scroll(ScrollMode)} using {@link Dialect#defaultScrollMode()}
 	 *
-	 * @return the result iterator
+	 * @apiNote The exact behavior of this method depends somewhat
+	 * on the JDBC driver's {@link java.sql.ResultSet} scrolling support
 	 */
-	ScrollableResults scroll();
+	ScrollableResults<R> scroll();
 
 	/**
-	 * Return the query results as ScrollableResults. The scrollability of the returned results
+	 * Returns scrollable access to the query results.  The capabilities of the
-	 * depends upon JDBC driver support for scrollable ResultSets.
+	 * returned ScrollableResults depend on the specified ScrollMode.
-	 *
-	 * @param scrollMode The scroll mode
-	 *
-	 * @return the result iterator
-	 *
-	 * @see ScrollableResults
-	 * @see ScrollMode
 	 *
+	 * @apiNote The exact behavior of this method depends somewhat
+	 * on the JDBC driver's {@link java.sql.ResultSet} scrolling support
 	 */
-	ScrollableResults scroll(ScrollMode scrollMode);
+	ScrollableResults<R> scroll(ScrollMode scrollMode);
 
 	/**
 	 * Return the query results as a <tt>List</tt>. If the query contains

hibernate-core/src/main/java/org/hibernate/query/internal/AbstractProducedQuery.java

@@ -11,7 +11,6 @@ import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZonedDateTime;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Calendar;
 import java.util.Collection;
@@ -44,6 +43,7 @@ import org.hibernate.PropertyNotFoundException;
 import org.hibernate.QueryParameterException;
 import org.hibernate.ScrollMode;
 import org.hibernate.TypeMismatchException;
+import org.hibernate.dialect.Dialect;
 import org.hibernate.engine.query.spi.EntityGraphQueryHint;
 import org.hibernate.engine.spi.ExceptionConverter;
 import org.hibernate.engine.spi.QueryParameters;
@@ -1390,51 +1390,25 @@ public abstract class AbstractProducedQuery<R> implements QueryImplementor<R> {
 	}
 
 	@Override
-	public ScrollableResultsImplementor scroll() {
+	public ScrollableResultsImplementor<R> scroll() {
-		return scroll( getSession().getJdbcServices().getJdbcEnvironment().getDialect().defaultScrollMode() );
+		final Dialect dialect = getSession().getJdbcServices().getJdbcEnvironment().getDialect();
+		return scroll( dialect.defaultScrollMode() );
 	}
 
 	@Override
-	public ScrollableResultsImplementor scroll(ScrollMode scrollMode) {
-		beforeQuery();
-		try {
-			return doScroll( scrollMode );
-		}
-		finally {
-			afterQuery();
-		}
-	}
-
-	protected ScrollableResultsImplementor doScroll(ScrollMode scrollMode) {
-		throw new NotYetImplementedFor6Exception( getClass() );
-
-//		if ( getMaxResults() == 0 ) {
-//			return EmptyScrollableResults.INSTANCE;
-//		}
-//
-//		final String query = getQueryParameterBindings().expandListValuedParameters( getQueryString(), getSession() );
-//		QueryParameters queryParameters = makeQueryParametersForExecution( query );
-//		queryParameters.setScrollMode( scrollMode );
-//		return getSession().scroll( query, queryParameters );
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
 	public Stream<R> stream() {
 		if (getMaxResults() == 0){
 			final Spliterator<R> spliterator = Spliterators.emptySpliterator();
 			return StreamSupport.stream( spliterator, false );
 		}
-		final ScrollableResultsImplementor scrollableResults = scroll( ScrollMode.FORWARD_ONLY );
+		final ScrollableResultsImplementor<R> scrollableResults = scroll( ScrollMode.FORWARD_ONLY );
 		final ScrollableResultsIterator<R> iterator = new ScrollableResultsIterator<>( scrollableResults );
 		final Spliterator<R> spliterator = Spliterators.spliteratorUnknownSize( iterator, Spliterator.NONNULL );
 
-		final Stream<R> stream = new StreamDecorator(
+		return new StreamDecorator<>(
 				StreamSupport.stream( spliterator, false ),
 				scrollableResults::close
 		);
-
-		return stream;
 	}
 
 	@Override

hibernate-core/src/main/java/org/hibernate/query/internal/QueryImpl.java

@@ -1,104 +0,0 @@
-/*
- * Hibernate, Relational Persistence for Idiomatic Java
- *
- * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
- * See the lgpl.txt file in the root directory or <http://www.gnu.org/licenses/lgpl-2.1.html>.
- */
-package org.hibernate.query.internal;
-
-import java.util.Collection;
-
-import org.hibernate.NotYetImplementedFor6Exception;
-import org.hibernate.engine.spi.SharedSessionContractImplementor;
-import org.hibernate.query.Query;
-import org.hibernate.query.ResultListTransformer;
-import org.hibernate.query.TupleTransformer;
-import org.hibernate.query.spi.ParameterMetadataImplementor;
-import org.hibernate.query.spi.QueryParameterBindings;
-
-/**
- * @author Steve Ebersole
- */
-public class QueryImpl<R> extends AbstractProducedQuery<R> implements Query<R> {
-	private final String queryString;
-
-	private final QueryParameterBindingsImpl queryParameterBindings;
-
-	public QueryImpl(
-			SharedSessionContractImplementor producer,
-			ParameterMetadataImplementor parameterMetadata,
-			String queryString) {
-		super( producer, parameterMetadata );
-		this.queryString = queryString;
-		this.queryParameterBindings = QueryParameterBindingsImpl.from(
-				parameterMetadata,
-				producer.getFactory(),
-				producer.isQueryParametersValidationEnabled()
-		);
-	}
-
-	@Override
-	protected QueryParameterBindings getQueryParameterBindings() {
-		return queryParameterBindings;
-	}
-
-	@Override
-	public String getQueryString() {
-		return queryString;
-	}
-
-	@Override
-	public Query<R> setTupleTransformer(TupleTransformer<R> transformer) {
-		throw new NotYetImplementedFor6Exception( getClass() );
-	}
-
-	@Override
-	public Query<R> setResultListTransformer(ResultListTransformer transformer) {
-		throw new NotYetImplementedFor6Exception( getClass() );
-	}
-
-	@Override
-	public Query<R> setParameterList(String name, Collection values, Class type) {
-		throw new NotYetImplementedFor6Exception( getClass() );
-	}
-
-	@Override
-	public Query<R> setParameterList(int position, Collection values, Class type) {
-		throw new NotYetImplementedFor6Exception( getClass() );
-	}
-
-	@Override
-	protected boolean isNativeQuery() {
-		return false;
-	}
-
-	@Override
-	public SharedSessionContractImplementor getSession() {
-		throw new NotYetImplementedFor6Exception( getClass() );
-	}
-
-	@Override
-	public QueryParameterBindings getParameterBindings() {
-		throw new NotYetImplementedFor6Exception( getClass() );
-	}
-
-//	@Override
-//	public Type[] getReturnTypes() {
-//		return getProducer().getFactory().getReturnTypes( queryString );
-//	}
-//
-//	@Override
-//	public String[] getReturnAliases() {
-//		return getProducer().getFactory().getReturnAliases( queryString );
-//	}
-//
-//	@Override
-//	public Query setEntity(int position, Object val) {
-//		return setParameter( position, val, getProducer().getFactory().getTypeHelper().entity( resolveEntityName( val ) ) );
-//	}
-//
-//	@Override
-//	public Query setEntity(String name, Object val) {
-//		return setParameter( name, val, getProducer().getFactory().getTypeHelper().entity( resolveEntityName( val ) ) );
-//	}
-}

hibernate-core/src/main/java/org/hibernate/query/spi/AbstractQuery.java

@@ -1425,23 +1425,10 @@ public abstract class AbstractQuery<R> implements QueryImplementor<R> {
 	}
 
 	@Override
-	public ScrollableResultsImplementor<?> scroll() {
+	public ScrollableResultsImplementor<R> scroll() {
 		return scroll( getSession().getFactory().getJdbcServices().getJdbcEnvironment().getDialect().defaultScrollMode() );
 	}
 
-	@Override
-	public ScrollableResultsImplementor<?> scroll(ScrollMode scrollMode) {
-		beforeQuery( false );
-		try {
-			return doScroll( scrollMode );
-		}
-		finally {
-			afterQuery();
-		}
-	}
-
-	protected abstract ScrollableResultsImplementor<?> doScroll(ScrollMode scrollMode);
-
 	@Override
 	@SuppressWarnings( {"unchecked", "rawtypes"} )
 	public Stream<R> stream() {

hibernate-core/src/main/java/org/hibernate/query/spi/QueryImplementor.java

@@ -9,6 +9,7 @@ package org.hibernate.query.spi;
 import java.io.Serializable;
 
 import org.hibernate.Incubating;
+import org.hibernate.ScrollMode;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.query.Query;
 
@@ -27,4 +28,10 @@ public interface QueryImplementor<R> extends Query<R> {
 	void setOptionalObject(Object optionalObject);
 
 	QueryParameterBindings getParameterBindings();
+
+	@Override
+	ScrollableResultsImplementor<R> scroll();
+
+	@Override
+	ScrollableResultsImplementor<R> scroll(ScrollMode scrollMode);
 }

hibernate-core/src/main/java/org/hibernate/query/sql/internal/NativeQueryImpl.java

@@ -539,7 +539,7 @@ public class NativeQueryImpl<R>
 	}
 
 	@Override
-	protected ScrollableResultsImplementor doScroll(ScrollMode scrollMode) {
+	public ScrollableResultsImplementor<R> scroll(ScrollMode scrollMode) {
 		return resolveSelectQueryPlan().performScroll( scrollMode, this );
 	}
 

hibernate-core/src/main/java/org/hibernate/query/sqm/internal/ConcreteSqmSelectQueryPlan.java

@@ -12,7 +12,6 @@ import java.util.Map;
 import javax.persistence.Tuple;
 import javax.persistence.TupleElement;
 
-import org.hibernate.NotYetImplementedFor6Exception;
 import org.hibernate.ScrollMode;
 import org.hibernate.engine.jdbc.env.spi.JdbcEnvironment;
 import org.hibernate.engine.jdbc.spi.JdbcServices;
@@ -33,8 +32,8 @@ import org.hibernate.query.sqm.tree.select.SqmSelectStatement;
 import org.hibernate.query.sqm.tree.select.SqmSelection;
 import org.hibernate.sql.ast.SqlAstTranslatorFactory;
 import org.hibernate.sql.ast.spi.FromClauseAccess;
-import org.hibernate.sql.exec.spi.ExecutionContext;
 import org.hibernate.sql.ast.tree.expression.JdbcParameter;
+import org.hibernate.sql.exec.spi.ExecutionContext;
 import org.hibernate.sql.exec.spi.JdbcParameterBindings;
 import org.hibernate.sql.exec.spi.JdbcSelect;
 import org.hibernate.sql.results.internal.RowTransformerJpaTupleImpl;
@@ -176,6 +175,40 @@ public class ConcreteSqmSelectQueryPlan<R> implements SelectQueryPlan<R> {
 		}
 	}
 
+
+
+
+
+	@Override
+	public ScrollableResultsImplementor<R> performScroll(ScrollMode scrollMode, ExecutionContext executionContext) {
+		final SharedSessionContractImplementor session = executionContext.getSession();
+
+		final CacheableSqmInterpretation sqmInterpretation = resolveCacheableSqmInterpretation( executionContext );
+
+		final JdbcParameterBindings jdbcParameterBindings = SqmUtil.createJdbcParameterBindings(
+				executionContext.getQueryParameterBindings(),
+				domainParameterXref,
+				sqmInterpretation.getJdbcParamsXref(),
+				session.getFactory().getDomainModel(),
+				sqmInterpretation.getTableGroupAccess()::findTableGroup,
+				session
+		);
+		sqmInterpretation.getJdbcSelect().bindFilterJdbcParameters( jdbcParameterBindings );
+
+		try {
+			return session.getFactory().getJdbcServices().getJdbcSelectExecutor().scroll(
+					sqmInterpretation.getJdbcSelect(),
+					scrollMode,
+					jdbcParameterBindings,
+					executionContext,
+					rowTransformer
+			);
+		}
+		finally {
+			domainParameterXref.clearExpansions();
+		}
+	}
+
 	private volatile CacheableSqmInterpretation cacheableSqmInterpretation;
 
 	private CacheableSqmInterpretation resolveCacheableSqmInterpretation(ExecutionContext executionContext) {
@@ -265,46 +298,4 @@ public class ConcreteSqmSelectQueryPlan<R> implements SelectQueryPlan<R> {
 			return jdbcParamsXref;
 		}
 	}
-
-
-
-
-
-	@Override
-	@SuppressWarnings("unchecked")
-	public ScrollableResultsImplementor performScroll(ScrollMode scrollMode, ExecutionContext executionContext) {
-		throw new NotYetImplementedFor6Exception( getClass() );
-
-//		final SqmSelectToSqlAstConverter sqmConverter = getSqmSelectToSqlAstConverter( executionContext );
-//
-//		final SqmSelectInterpretation interpretation = sqmConverter.interpret( sqm );
-//
-//		final JdbcSelect jdbcSelect = SqlAstSelectToJdbcSelectConverter.interpret(
-//				interpretation,
-//				executionContext.getSession().getSessionFactory()
-//		);
-//
-//		final Map<QueryParameterImplementor<?>, Map<SqmParameter, List<JdbcParameter>>> jdbcParamsXref =
-//				SqmConsumeHelper.generateJdbcParamsXref( domainParameterXref, sqmConverter );
-//
-//		final JdbcParameterBindings jdbcParameterBindings = QueryHelper.createJdbcParameterBindings(
-//				executionContext.getDomainParameterBindingContext().getQueryParameterBindings(),
-//				domainParameterXref,
-//				jdbcParamsXref,
-//				executionContext.getSession()
-//		);
-//
-//		try {
-//			return JdbcSelectExecutorStandardImpl.INSTANCE.scroll(
-//					jdbcSelect,
-//					scrollMode,
-//					jdbcParameterBindings,
-//					executionContext,
-//					rowTransformer
-//			);
-//		}
-//		finally {
-//			domainParameterXref.clearExpansions();
-//		}
-	}
 }

hibernate-core/src/main/java/org/hibernate/query/sqm/internal/QuerySqmImpl.java

@@ -517,7 +517,7 @@ public class QuerySqmImpl<R>
 	}
 
 	@Override
-	protected ScrollableResultsImplementor doScroll(ScrollMode scrollMode) {
+	public ScrollableResultsImplementor<R> scroll(ScrollMode scrollMode) {
 		SqmUtil.verifyIsSelectStatement( getSqmStatement() );
 		getSession().prepareForQueryExecution( requiresTxn( getLockOptions().findGreatestLockMode() ) );
 

hibernate-core/src/main/java/org/hibernate/sql/exec/internal/JdbcSelectExecutorStandardImpl.java

@@ -20,6 +20,8 @@ import org.hibernate.ScrollMode;
 import org.hibernate.cache.spi.QueryKey;
 import org.hibernate.cache.spi.QueryResultsCache;
 import org.hibernate.engine.spi.PersistenceContext;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.loader.ast.spi.AfterLoadAction;
 import org.hibernate.query.internal.ScrollableResultsIterator;
 import org.hibernate.query.spi.ScrollableResultsImplementor;
@@ -37,6 +39,7 @@ import org.hibernate.sql.results.jdbc.internal.JdbcValuesSourceProcessingStateSt
 import org.hibernate.sql.results.jdbc.internal.ResultSetAccess;
 import org.hibernate.sql.results.jdbc.spi.JdbcValues;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesMapping;
+import org.hibernate.sql.results.jdbc.spi.JdbcValuesMappingProducer;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
 import org.hibernate.sql.results.spi.ListResultsConsumer;
 import org.hibernate.sql.results.spi.ResultsConsumer;
@@ -62,11 +65,11 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 
 	@Override
 	public <R> List<R> list(
-		JdbcSelect jdbcSelect,
+			JdbcSelect jdbcSelect,
-		JdbcParameterBindings jdbcParameterBindings,
+			JdbcParameterBindings jdbcParameterBindings,
-		ExecutionContext executionContext,
+			ExecutionContext executionContext,
-		RowTransformer<R> rowTransformer,
+			RowTransformer<R> rowTransformer,
-		boolean uniqueFilter) {
+			boolean uniqueFilter) {
 		return executeQuery(
 				jdbcSelect,
 				jdbcParameterBindings,
@@ -162,6 +165,7 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 
 		final JdbcValues jdbcValues = resolveJdbcValuesSource(
 				jdbcSelect,
+				resultsConsumer.canResultsBeCached(),
 				executionContext,
 				new DeferredResultSetAccess(
 						jdbcSelect,
@@ -238,22 +242,27 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 	@SuppressWarnings("unchecked")
 	private JdbcValues resolveJdbcValuesSource(
 			JdbcSelect jdbcSelect,
+			boolean canBeCached,
 			ExecutionContext executionContext,
 			ResultSetAccess resultSetAccess) {
+		final SharedSessionContractImplementor session = executionContext.getSession();
+		final SessionFactoryImplementor factory = session.getFactory();
+		final boolean queryCacheEnabled = factory.getSessionFactoryOptions().isQueryCacheEnabled();
+
 		final List<Object[]> cachedResults;
 
-		final boolean queryCacheEnabled = executionContext.getSession().getFactory().getSessionFactoryOptions().isQueryCacheEnabled();
+
 		final CacheMode cacheMode = JdbcExecHelper.resolveCacheMode( executionContext );
 
-		final JdbcValuesMapping jdbcValuesMapping = jdbcSelect.getJdbcValuesMappingProducer()
+		final JdbcValuesMappingProducer mappingProducer = jdbcSelect.getJdbcValuesMappingProducer();
-				.resolve( resultSetAccess, executionContext.getSession().getFactory() );
+		final JdbcValuesMapping jdbcValuesMapping = mappingProducer.resolve( resultSetAccess, factory );
 
 		final QueryKey queryResultsCacheKey;
 
 		if ( queryCacheEnabled && cacheMode.isGetEnabled() ) {
 			SqlExecLogger.INSTANCE.debugf( "Reading Query result cache data per CacheMode#isGetEnabled [%s]", cacheMode.name() );
 
-			final QueryResultsCache queryCache = executionContext.getSession().getFactory()
+			final QueryResultsCache queryCache = factory
 					.getCache()
 					.getQueryResultsCache( executionContext.getQueryOptions().getResultCacheRegionName() );
 
@@ -267,7 +276,7 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 					jdbcSelect.getSql(),
 					executionContext.getQueryOptions().getLimit(),
 					executionContext.getQueryParameterBindings(),
-					executionContext.getSession()
+					session
 			);
 
 			cachedResults = queryCache.get(
@@ -276,7 +285,7 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 					// todo (6.0) : `querySpaces` and `session` make perfect sense as args, but its odd passing those into this method just to pass along
 					//		atm we do not even collect querySpaces, but we need to
 					jdbcSelect.getAffectedTableNames(),
-					executionContext.getSession()
+					session
 			);
 
 			// todo (6.0) : `querySpaces` and `session` are used in QueryCache#get to verify "up-to-dateness" via UpdateTimestampsCache
@@ -298,17 +307,14 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 		if ( cachedResults == null || cachedResults.isEmpty() ) {
 			return new JdbcValuesResultSetImpl(
 					resultSetAccess,
-					queryResultsCacheKey,
+					canBeCached ? queryResultsCacheKey : null,
 					executionContext.getQueryOptions(),
 					jdbcValuesMapping,
 					executionContext
 			);
 		}
 		else {
-			return new JdbcValuesCacheHit(
+			return new JdbcValuesCacheHit( cachedResults, jdbcValuesMapping );
-					cachedResults,
-					jdbcValuesMapping
-			);
 		}
 	}
 

hibernate-core/src/main/java/org/hibernate/sql/results/internal/RowProcessingStateStandardImpl.java

@@ -6,6 +6,7 @@
  */
 package org.hibernate.sql.results.internal;
 
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
 
@@ -38,6 +39,8 @@ public class RowProcessingStateStandardImpl implements RowProcessingState {
 
 	private final RowReader<?> rowReader;
 	private final JdbcValues jdbcValues;
+
+	// todo (6.0) : why doesn't this just use the array from JdbcValues?
 	private Object[] currentRowJdbcValues;
 
 	public RowProcessingStateStandardImpl(
@@ -70,7 +73,7 @@ public class RowProcessingStateStandardImpl implements RowProcessingState {
 		return rowReader;
 	}
 
-	public boolean next() throws SQLException {
+	public boolean next() {
 		if ( jdbcValues.next( this ) ) {
 			currentRowJdbcValues = jdbcValues.getCurrentRowValuesArray();
 			return true;
@@ -81,6 +84,65 @@ public class RowProcessingStateStandardImpl implements RowProcessingState {
 		}
 	}
 
+	public boolean previous() {
+		if ( jdbcValues.previous( this ) ) {
+			currentRowJdbcValues = jdbcValues.getCurrentRowValuesArray();
+			return true;
+		}
+		else {
+			currentRowJdbcValues = null;
+			return false;
+		}
+	}
+
+	public boolean scroll(int i) {
+		if ( jdbcValues.scroll( i, this ) ) {
+			currentRowJdbcValues = jdbcValues.getCurrentRowValuesArray();
+			return true;
+		}
+		else {
+			currentRowJdbcValues = null;
+			return false;
+		}
+	}
+
+	public boolean position(int i) {
+		if ( jdbcValues.position( i, this ) ) {
+			currentRowJdbcValues = jdbcValues.getCurrentRowValuesArray();
+			return true;
+		}
+		else {
+			currentRowJdbcValues = null;
+			return false;
+		}
+	}
+
+	public int getPosition() {
+		return jdbcValues.getPosition();
+	}
+
+	public boolean first() {
+		if ( jdbcValues.first( this ) ) {
+			currentRowJdbcValues = jdbcValues.getCurrentRowValuesArray();
+			return true;
+		}
+		else {
+			currentRowJdbcValues = null;
+			return false;
+		}
+	}
+
+	public boolean last() {
+		if ( jdbcValues.last( this ) ) {
+			currentRowJdbcValues = jdbcValues.getCurrentRowValuesArray();
+			return true;
+		}
+		else {
+			currentRowJdbcValues = null;
+			return false;
+		}
+	}
+
 	@Override
 	public Object getJdbcValue(int position) {
 		return currentRowJdbcValues[ position ];

hibernate-core/src/main/java/org/hibernate/sql/results/jdbc/internal/AbstractJdbcValues.java

@@ -6,8 +6,6 @@
  */
 package org.hibernate.sql.results.jdbc.internal;
 
-import java.sql.SQLException;
-
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.sql.results.caching.QueryCachePutManager;
 import org.hibernate.sql.results.jdbc.spi.JdbcValues;
@@ -27,10 +25,9 @@ public abstract class AbstractJdbcValues implements JdbcValues {
 	}
 
 	@Override
-	public final boolean next(RowProcessingState rowProcessingState) throws SQLException {
+	public final boolean next(RowProcessingState rowProcessingState) {
 		final boolean hadRow = processNext( rowProcessingState );
 		if ( hadRow ) {
-
 			queryCachePutManager.registerJdbcRow( getCurrentRowValuesArray() );
 		}
 		return hadRow;
@@ -38,6 +35,36 @@ public abstract class AbstractJdbcValues implements JdbcValues {
 
 	protected abstract boolean processNext(RowProcessingState rowProcessingState);
 
+	@Override
+	public boolean previous(RowProcessingState rowProcessingState) {
+		// NOTE : we do not even bother interacting with the query-cache put manager because
+		//		this method is implicitly related to scrolling and caching of scrolled results
+		//		is not supported
+		return processPrevious( rowProcessingState );
+	}
+
+	protected abstract boolean processPrevious(RowProcessingState rowProcessingState);
+
+	@Override
+	public boolean scroll(int numberOfRows, RowProcessingState rowProcessingState) {
+		// NOTE : we do not even bother interacting with the query-cache put manager because
+		//		this method is implicitly related to scrolling and caching of scrolled results
+		//		is not supported
+		return processScroll( numberOfRows, rowProcessingState );
+	}
+
+	protected abstract boolean processScroll(int numberOfRows, RowProcessingState rowProcessingState);
+
+	@Override
+	public boolean position(int position, RowProcessingState rowProcessingState) {
+		// NOTE : we do not even bother interacting with the query-cache put manager because
+		//		this method is implicitly related to scrolling and caching of scrolled results
+		//		is not supported
+		return processPosition( position, rowProcessingState );
+	}
+
+	protected abstract boolean processPosition(int position, RowProcessingState rowProcessingState);
+
 	@Override
 	public final void finishUp(SharedSessionContractImplementor session) {
 		queryCachePutManager.finishUp( session );

hibernate-core/src/main/java/org/hibernate/sql/results/jdbc/internal/JdbcValuesCacheHit.java

@@ -24,7 +24,7 @@ public class JdbcValuesCacheHit extends AbstractJdbcValues {
 
 	private Object[][] cachedData;
 	private final int numberOfRows;
-	private JdbcValuesMapping resolvedMapping;
+	private final JdbcValuesMapping resolvedMapping;
 	private int position = -1;
 
 	public JdbcValuesCacheHit(Object[][] cachedData, JdbcValuesMapping resolvedMapping) {
@@ -64,11 +64,112 @@ public class JdbcValuesCacheHit extends AbstractJdbcValues {
 		// NOTE : explicitly skipping limit handling because the cached state ought
 		// 		already be the limited size since the cache key includes limits
 
-		if ( position >= numberOfRows - 1 ) {
+		position++;
+
+		if ( position >= numberOfRows ) {
+			position = numberOfRows;
 			return false;
 		}
 
-		position++;
+		return true;
+	}
+
+	@Override
+	protected boolean processPrevious(RowProcessingState rowProcessingState) {
+		ResultsLogger.LOGGER.tracef( "JdbcValuesCacheHit#processPrevious : position = %i; numberOfRows = %i", position, numberOfRows );
+
+		// NOTE : explicitly skipping limit handling because the cached state ought
+		// 		already be the limited size since the cache key includes limits
+
+		position--;
+
+		if ( position >= numberOfRows ) {
+			position = numberOfRows;
+			return false;
+		}
+
+		return true;
+	}
+
+	@Override
+	protected boolean processScroll(int numberOfRows, RowProcessingState rowProcessingState) {
+		ResultsLogger.LOGGER.tracef( "JdbcValuesCacheHit#processScroll(%i) : position = %i; numberOfRows = %i", numberOfRows, position, this.numberOfRows );
+
+		// NOTE : explicitly skipping limit handling because the cached state should
+		// 		already be the limited size since the cache key includes limits
+
+		position += numberOfRows;
+
+		if ( position > this.numberOfRows ) {
+			position = this.numberOfRows;
+			return false;
+		}
+
+		return true;
+	}
+
+	@Override
+	public int getPosition() {
+		return position;
+	}
+
+	@Override
+	protected boolean processPosition(int position, RowProcessingState rowProcessingState) {
+		ResultsLogger.LOGGER.tracef( "JdbcValuesCacheHit#processPosition(%i) : position = %i; numberOfRows = %i", position, this.position, this.numberOfRows );
+
+		// NOTE : explicitly skipping limit handling because the cached state should
+		// 		already be the limited size since the cache key includes limits
+
+		if ( position < 0 ) {
+			// we need to subtract it from `numberOfRows`
+			final int newPosition = numberOfRows + position;
+			ResultsLogger.LOGGER.debugf(
+					"Translated negative absolute position `%i` into `%` based on `%i` number of rows",
+					position,
+					newPosition,
+					numberOfRows
+			);
+			position = newPosition;
+		}
+
+		if ( position > numberOfRows ) {
+			ResultsLogger.LOGGER.debugf(
+					"Absolute position `%i` exceeded number of rows `%i`",
+					position,
+					numberOfRows
+			);
+			this.position = numberOfRows;
+			return false;
+		}
+
+		this.position = position;
+		return true;
+	}
+
+	@Override
+	public boolean isBeforeFirst(RowProcessingState rowProcessingState) {
+		return position < 0;
+	}
+
+	@Override
+	public boolean first(RowProcessingState rowProcessingState) {
+		position = 0;
+		return numberOfRows > 0;
+	}
+
+	@Override
+	public boolean isAfterLast(RowProcessingState rowProcessingState) {
+		return position >= numberOfRows;
+	}
+
+	@Override
+	public boolean last(RowProcessingState rowProcessingState) {
+		if ( numberOfRows == 0 ) {
+			position = 0;
+			return false;
+		}
+
+		position = numberOfRows - 1;
 		return true;
 	}
 

hibernate-core/src/main/java/org/hibernate/sql/results/jdbc/internal/JdbcValuesResultSetImpl.java

@@ -11,7 +11,6 @@ import java.sql.SQLException;
 import org.hibernate.CacheMode;
 import org.hibernate.cache.spi.QueryKey;
 import org.hibernate.cache.spi.QueryResultsCache;
-import org.hibernate.query.Limit;
 import org.hibernate.query.spi.QueryOptions;
 import org.hibernate.sql.ast.spi.SqlSelection;
 import org.hibernate.sql.exec.ExecutionException;
@@ -37,15 +36,6 @@ public class JdbcValuesResultSetImpl extends AbstractJdbcValues {
 	private final SqlSelection[] sqlSelections;
 	private final Object[] currentRowJdbcValues;
 
-	// todo (6.0) - manage limit-based skips
-
-	private final int numberOfRowsToProcess;
-
-	// we start position at -1 prior to any next call so that the first next call
-	//		increments position to 0, which is the first row
-	private int position = -1;
-
-
 	public JdbcValuesResultSetImpl(
 			ResultSetAccess resultSetAccess,
 			QueryKey queryCacheKey,
@@ -57,25 +47,10 @@ public class JdbcValuesResultSetImpl extends AbstractJdbcValues {
 		this.valuesMapping = valuesMapping;
 		this.executionContext = executionContext;
 
-		// todo (6.0) : decide how to handle paged/limited results
-		this.numberOfRowsToProcess = interpretNumberOfRowsToProcess( queryOptions );
-
 		this.sqlSelections = valuesMapping.getSqlSelections().toArray( new SqlSelection[0] );
 		this.currentRowJdbcValues = new Object[ sqlSelections.length ];
 	}
 
-	private static int interpretNumberOfRowsToProcess(QueryOptions queryOptions) {
-		if ( queryOptions == null || queryOptions.getLimit() == null ) {
-			return -1;
-		}
-		final Limit limit = queryOptions.getLimit();
-		if ( limit.getMaxRows() == null ) {
-			return -1;
-		}
-
-		return limit.getMaxRows();
-	}
-
 	private static QueryCachePutManager resolveQueryCachePutManager(
 			ExecutionContext executionContext,
 			QueryOptions queryOptions,
@@ -100,30 +75,164 @@ public class JdbcValuesResultSetImpl extends AbstractJdbcValues {
 
 	@Override
 	protected final boolean processNext(RowProcessingState rowProcessingState) {
-		if ( numberOfRowsToProcess != -1 && position > numberOfRowsToProcess ) {
+		return advance(
-			// numberOfRowsToProcess != -1 means we had some limit, and
+				() -> {
-			//		position > numberOfRowsToProcess means we have exceeded the
+					try {
-			// 		number of limited rows
+						//noinspection RedundantIfStatement
+						if ( ! resultSetAccess.getResultSet().next() ) {
+							return false;
+						}
+
+						return true;
+					}
+					catch (SQLException e) {
+						throw makeExecutionException( "Error advancing (next) ResultSet position", e );
+					}
+				}
+		);
+	}
+
+	@Override
+	protected boolean processPrevious(RowProcessingState rowProcessingState) {
+		return advance(
+				() -> {
+					try {
+						//noinspection RedundantIfStatement
+						if ( ! resultSetAccess.getResultSet().previous() ) {
+							return false;
+						}
+						return true;
+					}
+					catch (SQLException e) {
+						throw makeExecutionException( "Error advancing (previous) ResultSet position", e );
+					}
+				}
+		);
+	}
+
+	@Override
+	protected boolean processScroll(int numberOfRows, RowProcessingState rowProcessingState) {
+		return advance(
+				() -> {
+					try {
+						//noinspection RedundantIfStatement
+						if ( ! resultSetAccess.getResultSet().relative( numberOfRows ) ) {
+							return false;
+						}
+
+						return true;
+					}
+					catch (SQLException e) {
+						throw makeExecutionException( "Error advancing (scroll) ResultSet position", e );
+					}
+				}
+		);
+	}
+
+	@Override
+	public int getPosition() {
+		try {
+			return resultSetAccess.getResultSet().getRow() - 1;
+		}
+		catch (SQLException e) {
+			throw makeExecutionException( "Error calling ResultSet#getRow", e );
+		}
+	}
+
+	@Override
+	protected boolean processPosition(int position, RowProcessingState rowProcessingState) {
+		return advance(
+				() -> {
+					try {
+						//noinspection RedundantIfStatement
+						if ( ! resultSetAccess.getResultSet().absolute( position ) ) {
+							return false;
+						}
+
+						return true;
+					}
+					catch (SQLException e) {
+						throw makeExecutionException( "Error advancing (scroll) ResultSet position", e );
+					}
+				}
+		);
+	}
+
+	@Override
+	public boolean isBeforeFirst(RowProcessingState rowProcessingState) {
+		try {
+			return resultSetAccess.getResultSet().isBeforeFirst();
+		}
+		catch (SQLException e) {
+			throw makeExecutionException( "Error calling ResultSet#isBeforeFirst", e );
+		}
+	}
+
+	@Override
+	public boolean first(RowProcessingState rowProcessingState) {
+		return advance(
+				() -> {
+					try {
+						//noinspection RedundantIfStatement
+						if ( ! resultSetAccess.getResultSet().first() ) {
+							return false;
+						}
+
+						return true;
+					}
+					catch (SQLException e) {
+						throw makeExecutionException( "Error advancing (first) ResultSet position", e );
+					}
+				}
+		);
+	}
+
+	@Override
+	public boolean isAfterLast(RowProcessingState rowProcessingState) {
+		try {
+			return resultSetAccess.getResultSet().isAfterLast();
+		}
+		catch (SQLException e) {
+			throw makeExecutionException( "Error calling ResultSet#isAfterLast", e );
+		}
+	}
+
+	@Override
+	public boolean last(RowProcessingState rowProcessingState) {
+		return advance(
+				() -> {
+					try {
+						//noinspection RedundantIfStatement
+						if ( ! resultSetAccess.getResultSet().last() ) {
+							return false;
+						}
+
+						return true;
+					}
+					catch (SQLException e) {
+						throw makeExecutionException( "Error advancing (last) ResultSet position", e );
+					}
+				}
+		);
+	}
+
+	@FunctionalInterface
+	private interface Advancer {
+		boolean advance();
+	}
+
+	private boolean advance(Advancer advancer) {
+		final boolean hasResult = advancer.advance();
+		if ( ! hasResult ) {
 			return false;
 		}
 
-		position++;
-
 		try {
-			if ( !resultSetAccess.getResultSet().next() ) {
+			readCurrentRowValues();
-				return false;
-			}
-		}
-		catch (SQLException e) {
-			throw makeExecutionException( "Error advancing JDBC ResultSet", e );
-		}
-
-		try {
-			readCurrentRowValues( rowProcessingState );
 			return true;
 		}
 		catch (SQLException e) {
-			throw makeExecutionException( "Error reading JDBC row values", e );
+			throw makeExecutionException( "Error reading ResultSet row values", e );
 		}
 	}
 
@@ -137,7 +246,7 @@ public class JdbcValuesResultSetImpl extends AbstractJdbcValues {
 		);
 	}
 
-	private void readCurrentRowValues(RowProcessingState rowProcessingState) throws SQLException {
+	private void readCurrentRowValues() throws SQLException {
 		for ( final SqlSelection sqlSelection : sqlSelections ) {
 			currentRowJdbcValues[ sqlSelection.getValuesArrayPosition() ] = sqlSelection.getJdbcValueExtractor().extract(
 					resultSetAccess.getResultSet(),

hibernate-core/src/main/java/org/hibernate/sql/results/jdbc/spi/JdbcValues.java

@@ -6,8 +6,6 @@
  */
 package org.hibernate.sql.results.jdbc.spi;
 
-import java.sql.SQLException;
-
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 
 /**
@@ -21,19 +19,45 @@ import org.hibernate.engine.spi.SharedSessionContractImplementor;
 public interface JdbcValues {
 	JdbcValuesMapping getValuesMapping();
 
-	// todo : ? - add ResultSet.previous() and ResultSet.absolute(int) style methods (to support ScrollableResults)?
+	/**
+	 * Advances the "cursor position" and returns a boolean indicating whether
+	 * there is a row available to read via {@link #getCurrentRowValuesArray()}.
+	 *
+	 * @return {@code true} if there are results
+	 */
+	boolean next(RowProcessingState rowProcessingState);
 
 	/**
-	 * Think JDBC's {@code ResultSet#next}.  Advances the "cursor position"
+	 * Advances the "cursor position" in reverse and returns a boolean indicating whether
-	 * and return a boolean indicating whether advancing positioned the
+	 * there is a row available to read via {@link #getCurrentRowValuesArray()}.
-	 * cursor beyond the set of available results.
 	 *
-	 * @return {@code true} indicates the call did not position the cursor beyond
+	 * @return {@code true} if there are results available
-	 * the available results ({@link #getCurrentRowValuesArray} will not return
-	 * null); false indicates we are now beyond the end of the available results
-	 * ({@link #getCurrentRowValuesArray} will return null)
 	 */
-	boolean next(RowProcessingState rowProcessingState) throws SQLException;
+	boolean previous(RowProcessingState rowProcessingState);
+
+	/**
+	 * Advances the "cursor position" the indicated number of rows and returns a boolean
+	 * indicating whether there is a row available to read via {@link #getCurrentRowValuesArray()}.
+	 *
+	 * @param numberOfRows The number of rows to advance.  This can also be negative meaning to
+	 * move in reverse
+	 *
+	 * @return {@code true} if there are results available
+	 */
+	boolean scroll(int numberOfRows, RowProcessingState rowProcessingState);
+
+	/**
+	 * Moves the "cursor position" to the specified position
+	 */
+	boolean position(int position, RowProcessingState rowProcessingState);
+
+	int getPosition();
+
+	boolean isBeforeFirst(RowProcessingState rowProcessingState);
+	boolean first(RowProcessingState rowProcessingState);
+
+	boolean isAfterLast(RowProcessingState rowProcessingState);
+	boolean last(RowProcessingState rowProcessingState);
 
 	/**
 	 * Get the JDBC values for the row currently positioned at within
@@ -45,11 +69,7 @@ public interface JdbcValues {
 	Object[] getCurrentRowValuesArray();
 
 	/**
-	 * todo (6.0) : is this needed?
+	 * Give implementations a chance to finish processing
-	 * 		^^ it's supposed to give impls a chance to write to the query cache
-	 * 		or release ResultSet it.  But that could technically be handled by the
-	 * 		case of `#next` returning false the first time.
-	 * @param session
 	 */
 	void finishUp(SharedSessionContractImplementor session);
 }

hibernate-core/src/main/java/org/hibernate/sql/results/spi/ListResultsConsumer.java

@@ -81,16 +81,15 @@ public class ListResultsConsumer<R> implements ResultsConsumer<List<R>, R> {
 			persistenceContext.initializeNonLazyCollections();
 			return results;
 		}
-		catch (SQLException e) {
-			throw session.getJdbcServices().getSqlExceptionHelper().convert(
-					e,
-					"Error processing return rows"
-			);
-		}
 		finally {
 			rowReader.finishUp( jdbcValuesSourceProcessingState );
 			jdbcValuesSourceProcessingState.finishUp();
 			jdbcValues.finishUp( session );
 		}
 	}
+
+	@Override
+	public boolean canResultsBeCached() {
+		return true;
+	}
 }

hibernate-core/src/main/java/org/hibernate/sql/results/spi/ResultsConsumer.java

@@ -7,8 +7,8 @@
 package org.hibernate.sql.results.spi;
 
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
-import org.hibernate.sql.results.jdbc.internal.JdbcValuesSourceProcessingStateStandardImpl;
 import org.hibernate.sql.results.internal.RowProcessingStateStandardImpl;
+import org.hibernate.sql.results.jdbc.internal.JdbcValuesSourceProcessingStateStandardImpl;
 import org.hibernate.sql.results.jdbc.spi.JdbcValues;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
 
@@ -23,4 +23,6 @@ public interface ResultsConsumer<T, R> {
 			JdbcValuesSourceProcessingStateStandardImpl jdbcValuesSourceProcessingState,
 			RowProcessingStateStandardImpl rowProcessingState,
 			RowReader<R> rowReader);
+
+	boolean canResultsBeCached();
 }

hibernate-core/src/main/java/org/hibernate/sql/results/spi/RowReader.java

@@ -40,7 +40,7 @@ public interface RowReader<R> {
 	 * todo (6.0) : JdbcValuesSourceProcessingOptions is available through RowProcessingState - why pass it in separately
 	 * 		should use one approach or the other
 	 */
-	R readRow(RowProcessingState processingState, JdbcValuesSourceProcessingOptions options) throws SQLException;
+	R readRow(RowProcessingState processingState, JdbcValuesSourceProcessingOptions options);
 
 	/**
 	 * Called at the end of processing all rows

hibernate-core/src/main/java/org/hibernate/sql/results/spi/ScrollableResultsConsumer.java

@@ -60,6 +60,11 @@ public class ScrollableResultsConsumer<R> implements ResultsConsumer<ScrollableR
 		}
 	}
 
+	@Override
+	public boolean canResultsBeCached() {
+		return false;
+	}
+
 	private boolean containsCollectionFetches(JdbcValuesMapping valuesMapping) {
 		return false;
 	}

hibernate-core/src/test/java/org/hibernate/orm/test/query/results/ResultListTest.java

@@ -0,0 +1,202 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.orm.test.query.results;
+
+import java.util.List;
+import java.util.function.Consumer;
+import javax.persistence.Tuple;
+import javax.persistence.TupleElement;
+
+import org.hibernate.query.Query;
+import org.hibernate.query.spi.QueryImplementor;
+
+import org.hibernate.testing.orm.domain.gambit.BasicEntity;
+import org.hibernate.testing.orm.junit.DomainModel;
+import org.hibernate.testing.orm.junit.SessionFactory;
+import org.hibernate.testing.orm.junit.SessionFactoryScope;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.typeCompatibleWith;
+import static org.hibernate.orm.test.query.results.ScalarQueries.MULTI_SELECTION_QUERY;
+import static org.hibernate.orm.test.query.results.ScalarQueries.SINGLE_ALIASED_SELECTION_QUERY;
+import static org.hibernate.orm.test.query.results.ScalarQueries.SINGLE_SELECTION_QUERY;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Tests of the Query's "domain results" via normal `Query#list` operations
+ */
+@DomainModel( annotatedClasses = BasicEntity.class )
+@SessionFactory
+public class ResultListTest {
+
+	@BeforeEach
+	public void setUpTestData(SessionFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> session.persist( new BasicEntity( 1, "value" ) )
+		);
+	}
+
+	@AfterEach
+	public void cleanUpTestData(SessionFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> session.createQuery( "delete BasicEntity" ).executeUpdate()
+		);
+	}
+
+	@Test
+	public void testSelectionTupleList(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Tuple> query = session.createQuery( SINGLE_SELECTION_QUERY, Tuple.class );
+					verifyList(
+							query,
+							(tuple) -> {
+								assertThat( tuple.getElements().size(), is( 1 ) );
+								final TupleElement<?> element = tuple.getElements().get( 0 );
+								assertThat( element.getJavaType(), typeCompatibleWith( String.class ) );
+								assertThat( element.getAlias(), nullValue() );
+
+								assertThat( tuple.toArray().length, is( 1 ) );
+
+								final Object byPosition = tuple.get( 0 );
+								assertThat( byPosition, is( "value" ) );
+
+								try {
+									tuple.get( "data" );
+									fail( "Expecting IllegalArgumentException per JPA spec" );
+								}
+								catch (IllegalArgumentException e) {
+									// expected outcome
+								}
+							}
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testAliasedSelectionTupleList(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Tuple> query = session.createQuery( SINGLE_ALIASED_SELECTION_QUERY, Tuple.class );
+					verifyList(
+							query,
+							(tuple) -> {
+								assertThat( tuple.getElements().size(), is( 1 ) );
+								final TupleElement<?> element = tuple.getElements().get( 0 );
+								assertThat( element.getJavaType(), typeCompatibleWith( String.class ) );
+								assertThat( element.getAlias(), is( "state" ) );
+
+								assertThat( tuple.toArray().length, is( 1 ) );
+
+								final Object byPosition = tuple.get( 0 );
+								assertThat( byPosition, is( "value" ) );
+
+								final Object byName = tuple.get( "state" );
+								assertThat( byName, is( "value" ) );
+							}
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testSelectionsTupleList(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Tuple> query = session.createQuery( MULTI_SELECTION_QUERY, Tuple.class );
+					verifyList(
+							query,
+							(tuple) -> {
+								assertThat( tuple.getElements().size(), is( 2 ) );
+
+								{
+									final TupleElement<?> element = tuple.getElements().get( 0 );
+									assertThat( element.getJavaType(), typeCompatibleWith( Integer.class ) );
+									assertThat( element.getAlias(), nullValue() );
+								}
+
+								{
+									final TupleElement<?> element = tuple.getElements().get( 1 );
+									assertThat( element.getJavaType(), typeCompatibleWith( String.class ) );
+									assertThat( element.getAlias(), nullValue() );
+								}
+
+								assertThat( tuple.toArray().length, is( 2 ) );
+
+								{
+									final Object byPosition = tuple.get( 0 );
+									assertThat( byPosition, is( 1 ) );
+
+									try {
+										tuple.get( "id" );
+										fail( "Expecting IllegalArgumentException per JPA spec" );
+									}
+									catch (IllegalArgumentException e) {
+										// expected outcome
+									}
+								}
+
+								{
+									final Object byPosition = tuple.get( 1 );
+									assertThat( byPosition, is( "value" ) );
+
+									try {
+										tuple.get( "data" );
+										fail( "Expecting IllegalArgumentException per JPA spec" );
+									}
+									catch (IllegalArgumentException e) {
+										// expected outcome
+									}
+								}
+							}
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testSelectionList(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<String> query = session.createQuery( SINGLE_SELECTION_QUERY, String.class );
+					verifyList(
+							query,
+							(data) -> assertThat( data, is( "value" ) )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testScrollSelections(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Object[]> query = session.createQuery( MULTI_SELECTION_QUERY, Object[].class );
+					verifyList(
+							query,
+							(values) -> {
+								assertThat( values[0], is( 1 ) );
+								assertThat( values[1], is( "value" ) );
+							}
+					);
+				}
+		);
+	}
+
+
+	private static <R> void verifyList(Query<R> query, Consumer<R> validator) {
+		final List<R> results = query.list();
+		assertThat( results.size(), is( 1 ) );
+		validator.accept( results.get( 0 ) );
+	}
+}

hibernate-core/src/test/java/org/hibernate/orm/test/query/results/ScalarQueries.java

@@ -0,0 +1,16 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.orm.test.query.results;
+
+/**
+ * @author Steve Ebersole
+ */
+public interface ScalarQueries {
+	String SINGLE_SELECTION_QUERY = "select e.data from BasicEntity e order by e.data";
+	String MULTI_SELECTION_QUERY = "select e.id, e.data from BasicEntity e order by e.id";
+	String SINGLE_ALIASED_SELECTION_QUERY = "select e.data as state from BasicEntity e order by e.data";
+}

hibernate-core/src/test/java/org/hibernate/orm/test/query/results/ScrollableResultsTests.java

@@ -0,0 +1,281 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.orm.test.query.results;
+
+import java.util.function.Consumer;
+import javax.persistence.Tuple;
+import javax.persistence.TupleElement;
+
+import org.hibernate.ScrollMode;
+import org.hibernate.ScrollableResults;
+import org.hibernate.query.Query;
+import org.hibernate.query.spi.QueryImplementor;
+import org.hibernate.query.spi.ScrollableResultsImplementor;
+
+import org.hibernate.testing.orm.domain.gambit.BasicEntity;
+import org.hibernate.testing.orm.junit.DomainModel;
+import org.hibernate.testing.orm.junit.SessionFactory;
+import org.hibernate.testing.orm.junit.SessionFactoryScope;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static org.hamcrest.Matchers.typeCompatibleWith;
+import static org.hibernate.orm.test.query.results.ScalarQueries.MULTI_SELECTION_QUERY;
+import static org.hibernate.orm.test.query.results.ScalarQueries.SINGLE_ALIASED_SELECTION_QUERY;
+import static org.hibernate.orm.test.query.results.ScalarQueries.SINGLE_SELECTION_QUERY;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Tests of the Query's "domain results" via ScrollableResults
+ */
+@DomainModel( annotatedClasses = BasicEntity.class )
+@SessionFactory
+public class ScrollableResultsTests {
+
+	@BeforeEach
+	public void setUpTestData(SessionFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> session.persist( new BasicEntity( 1, "value" ) )
+		);
+	}
+
+	@AfterEach
+	public void cleanUpTestData(SessionFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> session.createQuery( "delete BasicEntity" ).executeUpdate()
+		);
+	}
+
+	@Test
+	public void testCursorPositioning(SessionFactoryScope scope) {
+		// create an extra row so we can better test cursor positioning
+		scope.inTransaction(
+				session -> session.persist( new BasicEntity( 2, "other" ) )
+		);
+
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<String> query = session.createQuery( SINGLE_SELECTION_QUERY, String.class );
+					final ScrollableResultsImplementor<String> results = query.scroll( ScrollMode.SCROLL_INSENSITIVE );
+
+					// try to initially read in reverse - should be false
+					assertThat( results.previous(), is( false ) );
+
+					// position at the first row
+					assertThat( results.next(), is( true ) );
+					String data = results.get();
+					assertThat( data, is( "other" ) );
+
+					// position at the second (last) row
+					assertThat( results.next(), is( true ) );
+					data = results.get();
+					assertThat( data, is( "value" ) );
+
+					// position after the second (last) row
+					assertThat( results.next(), is( false ) );
+
+					// position back to the second row
+					assertThat( results.previous(), is( true ) );
+					data = results.get();
+					assertThat( data, is( "value" ) );
+
+					// position back to the first row
+					assertThat( results.previous(), is( true ) );
+					data = results.get();
+					assertThat( data, is( "other" ) );
+
+					// position before the first row
+					assertThat( results.previous(), is( false ) );
+					assertThat( results.previous(), is( false ) );
+
+					assertThat( results.last(), is( true ) );
+					data = results.get();
+					assertThat( data, is( "value" ) );
+
+					assertThat( results.first(), is( true ) );
+					data = results.get();
+					assertThat( data, is( "other" ) );
+
+					assertThat( results.scroll( 1 ), is( true ) );
+					data = results.get();
+					assertThat( data, is( "value" ) );
+
+					assertThat( results.scroll( -1 ), is( true ) );
+					data = results.get();
+					assertThat( data, is( "other" ) );
+
+				}
+		);
+	}
+
+	@Test
+	public void testScrollSelectionTuple(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Tuple> query = session.createQuery( SINGLE_SELECTION_QUERY, Tuple.class );
+					verifyScroll(
+							query,
+							(tuple) -> {
+								assertThat( tuple.getElements().size(), is( 1 ) );
+								final TupleElement<?> element = tuple.getElements().get( 0 );
+								assertThat( element.getJavaType(), typeCompatibleWith( String.class ) );
+								assertThat( element.getAlias(), nullValue() );
+
+								assertThat( tuple.toArray().length, is( 1 ) );
+
+								final Object byPosition = tuple.get( 0 );
+								assertThat( byPosition, is( "value" ) );
+
+								try {
+									tuple.get( "data" );
+									fail( "Expecting IllegalArgumentException per JPA spec" );
+								}
+								catch (IllegalArgumentException e) {
+									// expected outcome
+								}
+							}
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testScrollAliasedSelectionTuple(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Tuple> query = session.createQuery( SINGLE_ALIASED_SELECTION_QUERY, Tuple.class );
+					verifyScroll(
+							query,
+							(tuple) -> {
+								assertThat( tuple.getElements().size(), is( 1 ) );
+								final TupleElement<?> element = tuple.getElements().get( 0 );
+								assertThat( element.getJavaType(), typeCompatibleWith( String.class ) );
+								assertThat( element.getAlias(), is( "state" ) );
+
+								assertThat( tuple.toArray().length, is( 1 ) );
+
+								final Object byPosition = tuple.get( 0 );
+								assertThat( byPosition, is( "value" ) );
+
+								final Object byName = tuple.get( "state" );
+								assertThat( byName, is( "value" ) );
+							}
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testScrollSelectionsTuple(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Tuple> query = session.createQuery( MULTI_SELECTION_QUERY, Tuple.class );
+					verifyScroll(
+							query,
+							(tuple) -> {
+								assertThat( tuple.getElements().size(), is( 2 ) );
+
+								{
+									final TupleElement<?> element = tuple.getElements().get( 0 );
+									assertThat( element.getJavaType(), typeCompatibleWith( Integer.class ) );
+									assertThat( element.getAlias(), nullValue() );
+								}
+
+								{
+									final TupleElement<?> element = tuple.getElements().get( 1 );
+									assertThat( element.getJavaType(), typeCompatibleWith( String.class ) );
+									assertThat( element.getAlias(), nullValue() );
+								}
+
+								assertThat( tuple.toArray().length, is( 2 ) );
+
+								{
+									final Object byPosition = tuple.get( 0 );
+									assertThat( byPosition, is( 1 ) );
+
+									try {
+										tuple.get( "id" );
+										fail( "Expecting IllegalArgumentException per JPA spec" );
+									}
+									catch (IllegalArgumentException e) {
+										// expected outcome
+									}
+								}
+
+								{
+									final Object byPosition = tuple.get( 1 );
+									assertThat( byPosition, is( "value" ) );
+
+									try {
+										tuple.get( "data" );
+										fail( "Expecting IllegalArgumentException per JPA spec" );
+									}
+									catch (IllegalArgumentException e) {
+										// expected outcome
+									}
+								}
+							}
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testScrollSelection(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<String> query = session.createQuery( SINGLE_SELECTION_QUERY, String.class );
+					verifyScroll(
+							query,
+							(data) -> assertThat( data, is( "value" ) )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testScrollSelections(SessionFactoryScope scope) {
+		scope.inTransaction(
+				session -> {
+					final QueryImplementor<Object[]> query = session.createQuery( MULTI_SELECTION_QUERY, Object[].class );
+					verifyScroll(
+							query,
+							(values) -> {
+								assertThat( values[0], is( 1 ) );
+								assertThat( values[1], is( "value" ) );
+							}
+					);
+				}
+		);
+	}
+
+	private static <R> void verifyScroll(Query<R> query, Consumer<R> validator) {
+		try ( final ScrollableResults<R> results = query.scroll( ScrollMode.FORWARD_ONLY ) ) {
+			assertThat( results.next(), is( true ) );
+			validator.accept( results.get() );
+		}
+
+		try ( final ScrollableResults<R> results = query.scroll( ScrollMode.SCROLL_INSENSITIVE ) ) {
+			assertThat( results.next(), is( true ) );
+			validator.accept( results.get() );
+		}
+
+		try ( final ScrollableResults<R> results = query.scroll( ScrollMode.SCROLL_SENSITIVE ) ) {
+			assertThat( results.next(), is( true ) );
+			validator.accept( results.get() );
+		}
+
+		try ( final ScrollableResults<R> results = query.scroll( ScrollMode.SCROLL_INSENSITIVE ) ) {
+			assertThat( results.next(), is( true ) );
+			validator.accept( results.get() );
+		}
+	}
+}