HHH-17226 Generify result retrieval in JdbcSelectExecutor, SelectQueryPlan and SqmSelectionQuery

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

@@ -95,7 +95,6 @@ public List<T> multiLoad(Object... ids) {
 			}
 		}
 
-		session.autoFlushIfRequired( (Set) CollectionHelper.setOf( entityDescriptor.getQuerySpaces() ) );
 		final LoadQueryInfluencers loadQueryInfluencers = session.getLoadQueryInfluencers();
 
 		try {
@@ -112,6 +111,7 @@ public List<T> multiLoad(Object... ids) {
 			}
 
 			try {
+				session.autoFlushIfRequired( (Set) CollectionHelper.setOf( entityDescriptor.getQuerySpaces() ) );
 				return (List<T>) entityDescriptor.getMultiNaturalIdLoader().multiLoad( ids, this, session );
 			}
 			finally {

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

@@ -11,6 +11,7 @@
 import org.hibernate.Incubating;
 import org.hibernate.ScrollMode;
 import org.hibernate.query.Query;
+import org.hibernate.sql.results.spi.ResultsConsumer;
 
 /**
  * General contract for performing execution of a query returning results.  These
@@ -32,6 +33,12 @@
  */
 @Incubating
 public interface SelectQueryPlan<R> extends QueryPlan {
+	/**
+	 * Execute the query
+	 *
+	 * @since 6.4
+	 */
+	<T> T executeQuery(DomainQueryExecutionContext executionContext, ResultsConsumer<T, R> resultsConsumer);
 	/**
 	 * Perform (execute) the query returning a List
 	 */

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

@@ -29,6 +29,7 @@
 import org.hibernate.sql.exec.spi.JdbcParameterBindings;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesMappingProducer;
 import org.hibernate.sql.results.spi.ListResultsConsumer;
+import org.hibernate.sql.results.spi.ResultsConsumer;
 
 /**
  * @author Steve Ebersole
@@ -61,6 +62,47 @@ public NativeSelectQueryPlanImpl(
 		this.affectedTableNames = affectedTableNames;
 	}
 
+	@Override
+	public <T> T executeQuery(DomainQueryExecutionContext executionContext, ResultsConsumer<T, R> resultsConsumer) {
+		final List<JdbcParameterBinder> jdbcParameterBinders;
+		final JdbcParameterBindings jdbcParameterBindings;
+
+		final QueryParameterBindings queryParameterBindings = executionContext.getQueryParameterBindings();
+		if ( parameterList == null || parameterList.isEmpty() ) {
+			jdbcParameterBinders = Collections.emptyList();
+			jdbcParameterBindings = JdbcParameterBindings.NO_BINDINGS;
+		}
+		else {
+			jdbcParameterBinders = new ArrayList<>( parameterList.size() );
+			jdbcParameterBindings = new JdbcParameterBindingsImpl(
+					queryParameterBindings,
+					parameterList,
+					jdbcParameterBinders,
+					executionContext.getSession().getFactory()
+			);
+		}
+
+		final JdbcOperationQuerySelect jdbcSelect = new JdbcOperationQuerySelect(
+				sql,
+				jdbcParameterBinders,
+				resultSetMapping,
+				affectedTableNames
+		);
+
+		return executionContext.getSession().getJdbcServices().getJdbcSelectExecutor().executeQuery(
+				jdbcSelect,
+				jdbcParameterBindings,
+				SqmJdbcExecutionContextAdapter.usingLockingAndPaging( executionContext ),
+				null,
+				null,
+				sqlString -> executionContext.getSession()
+						.getJdbcCoordinator()
+						.getStatementPreparer()
+						.prepareQueryStatement( sqlString, false, null ),
+				resultsConsumer
+		);
+	}
+
 	@Override
 	public List<R> performList(DomainQueryExecutionContext executionContext) {
 		final QueryOptions queryOptions = executionContext.getQueryOptions();
@@ -85,8 +127,6 @@ public List<R> performList(DomainQueryExecutionContext executionContext) {
 			);
 		}
 
-		executionContext.getSession().autoFlushIfRequired( affectedTableNames );
-
 		final JdbcOperationQuerySelect jdbcSelect = new JdbcOperationQuerySelect(
 				sql,
 				jdbcParameterBinders,
@@ -94,6 +134,7 @@ public List<R> performList(DomainQueryExecutionContext executionContext) {
 				affectedTableNames
 		);
 
+		executionContext.getSession().autoFlushIfRequired( jdbcSelect.getAffectedTableNames() );
 		return executionContext.getSession().getJdbcServices().getJdbcSelectExecutor().list(
 				jdbcSelect,
 				jdbcParameterBindings,
@@ -136,6 +177,7 @@ public ScrollableResultsImplementor<R> performScroll(ScrollMode scrollMode, Doma
 				affectedTableNames
 		);
 
+		executionContext.getSession().autoFlushIfRequired( jdbcSelect.getAffectedTableNames() );
 		return executionContext.getSession().getJdbcServices().getJdbcSelectExecutor().scroll(
 				jdbcSelect,
 				scrollMode,

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

@@ -16,6 +16,7 @@
 import org.hibernate.query.spi.Limit;
 import org.hibernate.query.spi.ScrollableResultsImplementor;
 import org.hibernate.query.spi.SelectQueryPlan;
+import org.hibernate.sql.results.spi.ResultsConsumer;
 
 /**
  * @author Steve Ebersole
@@ -27,6 +28,11 @@ public AggregatedSelectQueryPlanImpl(SelectQueryPlan<R>[] aggregatedQueryPlans)
 		this.aggregatedQueryPlans = aggregatedQueryPlans;
 	}
 
+	@Override
+	public <T> T executeQuery(DomainQueryExecutionContext executionContext, ResultsConsumer<T, R> resultsConsumer) {
+		throw new UnsupportedOperationException();
+	}
+
 	@Override
 	public List<R> performList(DomainQueryExecutionContext executionContext) {
 		final Limit effectiveLimit = executionContext.getQueryOptions().getEffectiveLimit();

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

@@ -53,6 +53,7 @@
 import org.hibernate.sql.results.internal.RowTransformerTupleTransformerAdapter;
 import org.hibernate.sql.results.internal.TupleMetadata;
 import org.hibernate.sql.results.spi.ListResultsConsumer;
+import org.hibernate.sql.results.spi.ResultsConsumer;
 import org.hibernate.sql.results.spi.RowTransformer;
 
 import static org.hibernate.internal.util.ReflectHelper.isClass;
@@ -69,6 +70,7 @@ public class ConcreteSqmSelectQueryPlan<R> implements SelectQueryPlan<R> {
 	private final SqmSelectStatement<?> sqm;
 	private final DomainParameterXref domainParameterXref;
 	private final RowTransformer<R> rowTransformer;
+	private final SqmInterpreter<Object, ResultsConsumer<?, R>> executeQueryInterpreter;
 	private final SqmInterpreter<List<R>, Void> listInterpreter;
 	private final SqmInterpreter<ScrollableResultsImplementor<R>, ScrollMode> scrollInterpreter;
 
@@ -93,6 +95,34 @@ public ConcreteSqmSelectQueryPlan(
 		else {
 			uniqueSemantic = ListResultsConsumer.UniqueSemantic.ALLOW;
 		}
+		this.executeQueryInterpreter = (resultsConsumer, executionContext, sqmInterpretation, jdbcParameterBindings) -> {
+			final SharedSessionContractImplementor session = executionContext.getSession();
+			final JdbcOperationQuerySelect jdbcSelect = sqmInterpretation.getJdbcSelect();
+			try {
+				final SubselectFetch.RegistrationHandler subSelectFetchKeyHandler = SubselectFetch.createRegistrationHandler(
+						session.getPersistenceContext().getBatchFetchQueue(),
+						sqmInterpretation.selectStatement,
+						JdbcParametersList.empty(),
+						jdbcParameterBindings
+				);
+
+				return session.getFactory().getJdbcServices().getJdbcSelectExecutor().executeQuery(
+						jdbcSelect,
+						jdbcParameterBindings,
+						listInterpreterExecutionContext( hql, executionContext, jdbcSelect, subSelectFetchKeyHandler ),
+						rowTransformer,
+						null,
+						sql -> executionContext.getSession()
+								.getJdbcCoordinator()
+								.getStatementPreparer()
+								.prepareQueryStatement( sql, false, null ),
+						resultsConsumer
+				);
+			}
+			finally {
+				domainParameterXref.clearExpansions();
+			}
+		};
 		this.listInterpreter = (unused, executionContext, sqmInterpretation, jdbcParameterBindings) -> {
 			final SharedSessionContractImplementor session = executionContext.getSession();
 			final JdbcOperationQuerySelect jdbcSelect = sqmInterpretation.getJdbcSelect();
@@ -105,7 +135,6 @@ public ConcreteSqmSelectQueryPlan(
 				);
 
 				session.autoFlushIfRequired( jdbcSelect.getAffectedTableNames() );
-
 				return session.getFactory().getJdbcServices().getJdbcSelectExecutor().list(
 						jdbcSelect,
 						jdbcParameterBindings,
@@ -120,6 +149,8 @@ public ConcreteSqmSelectQueryPlan(
 		};
 
 		this.scrollInterpreter = (scrollMode, executionContext, sqmInterpretation, jdbcParameterBindings) -> {
+			final SharedSessionContractImplementor session = executionContext.getSession();
+			final JdbcOperationQuerySelect jdbcSelect = sqmInterpretation.getJdbcSelect();
 			try {
 //				final SubselectFetch.RegistrationHandler subSelectFetchKeyHandler = SubselectFetch.createRegistrationHandler(
 //						executionContext.getSession().getPersistenceContext().getBatchFetchQueue(),
@@ -128,15 +159,15 @@ public ConcreteSqmSelectQueryPlan(
 //						jdbcParameterBindings
 //				);
 
-				final JdbcSelectExecutor jdbcSelectExecutor = executionContext.getSession()
+				final JdbcSelectExecutor jdbcSelectExecutor = session.getFactory()
-						.getFactory()
 						.getJdbcServices()
 						.getJdbcSelectExecutor();
+				session.autoFlushIfRequired( jdbcSelect.getAffectedTableNames() );
 				return jdbcSelectExecutor.scroll(
-						sqmInterpretation.getJdbcSelect(),
+						jdbcSelect,
 						scrollMode,
 						jdbcParameterBindings,
-						new SqmJdbcExecutionContextAdapter( executionContext, sqmInterpretation.jdbcSelect ),
+						new SqmJdbcExecutionContextAdapter( executionContext, jdbcSelect ),
 						rowTransformer
 				);
 			}
@@ -238,6 +269,16 @@ private static <T> RowTransformer<T> makeRowTransformerTupleTransformerAdapter(
 		);
 	}
 
+	@Override
+	public <T> T executeQuery(DomainQueryExecutionContext executionContext, ResultsConsumer<T, R> resultsConsumer) {
+		//noinspection unchecked,rawtypes
+		return withCacheableSqmInterpretation(
+				executionContext,
+				resultsConsumer,
+				(SqmInterpreter<T, ResultsConsumer<T, R>>) (SqmInterpreter) executeQueryInterpreter
+		);
+	}
+
 	@Override
 	public List<R> performList(DomainQueryExecutionContext executionContext) {
 		if ( executionContext.getQueryOptions().getEffectiveLimit().getMaxRowsJpa() == 0 ) {

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

@@ -57,6 +57,7 @@
 import org.hibernate.query.spi.SelectQueryPlan;
 import org.hibernate.query.sqm.SqmSelectionQuery;
 import org.hibernate.query.sqm.internal.SqmInterpretationsKey.InterpretationsKeySource;
+import org.hibernate.query.sqm.spi.SqmSelectionQueryImplementor;
 import org.hibernate.query.sqm.tree.SqmCopyContext;
 import org.hibernate.query.sqm.tree.expression.JpaCriteriaParameter;
 import org.hibernate.query.sqm.tree.expression.SqmJpaCriteriaParameterWrapper;
@@ -64,6 +65,7 @@
 import org.hibernate.query.sqm.tree.select.SqmSelectStatement;
 import org.hibernate.query.sqm.tree.select.SqmSelection;
 import org.hibernate.sql.results.internal.TupleMetadata;
+import org.hibernate.sql.results.spi.ResultsConsumer;
 import org.hibernate.type.descriptor.java.JavaType;
 
 import static java.util.stream.Collectors.toList;
@@ -85,7 +87,7 @@
  * @author Steve Ebersole
  */
 public class SqmSelectionQueryImpl<R> extends AbstractSelectionQuery<R>
-		implements SqmSelectionQuery<R>, InterpretationsKeySource {
+		implements SqmSelectionQueryImplementor<R>, InterpretationsKeySource {
 	private final String hql;
 	private SqmSelectStatement<R> sqm;
 
@@ -396,6 +398,10 @@ protected ScrollableResultsImplementor<R> doScroll(ScrollMode scrollMode) {
 		return resolveQueryPlan().performScroll( scrollMode, this );
 	}
 
+	@Override
+	public <T> T executeQuery(ResultsConsumer<T, R> resultsConsumer) {
+		return resolveQueryPlan().executeQuery( this, resultsConsumer );
+	}
 
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// Query plan

hibernate-core/src/main/java/org/hibernate/query/sqm/spi/DelegatingSqmSelectionQueryImplementor.java

@@ -0,0 +1,609 @@
+/*
+ * 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.sqm.spi;
+
+import java.time.Instant;
+import java.util.Calendar;
+import java.util.Collection;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import org.hibernate.CacheMode;
+import org.hibernate.FlushMode;
+import org.hibernate.Incubating;
+import org.hibernate.LockMode;
+import org.hibernate.LockOptions;
+import org.hibernate.Remove;
+import org.hibernate.ScrollMode;
+import org.hibernate.ScrollableResults;
+import org.hibernate.graph.GraphSemantic;
+import org.hibernate.query.BindableType;
+import org.hibernate.query.Order;
+import org.hibernate.query.Page;
+import org.hibernate.query.ParameterMetadata;
+import org.hibernate.query.QueryParameter;
+import org.hibernate.query.spi.QueryOptions;
+import org.hibernate.query.sqm.tree.SqmStatement;
+import org.hibernate.sql.results.spi.ResultsConsumer;
+
+import jakarta.persistence.CacheRetrieveMode;
+import jakarta.persistence.CacheStoreMode;
+import jakarta.persistence.EntityGraph;
+import jakarta.persistence.FlushModeType;
+import jakarta.persistence.LockModeType;
+import jakarta.persistence.Parameter;
+import jakarta.persistence.TemporalType;
+
+@Incubating
+public abstract class DelegatingSqmSelectionQueryImplementor<R> implements SqmSelectionQueryImplementor<R> {
+
+	protected abstract SqmSelectionQueryImplementor<R> getDelegate();
+
+	@Override
+	public FlushModeType getFlushMode() {
+		return getDelegate().getFlushMode();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setFlushMode(FlushModeType flushMode) {
+		getDelegate().setFlushMode( flushMode );
+		return this;
+	}
+
+	@Override
+	public FlushMode getHibernateFlushMode() {
+		return getDelegate().getHibernateFlushMode();
+	}
+
+	@Override
+	public Integer getTimeout() {
+		return getDelegate().getTimeout();
+	}
+
+	@Override
+	public String getComment() {
+		return getDelegate().getComment();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setComment(String comment) {
+		getDelegate().setComment( comment );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setHint(String hintName, Object value) {
+		getDelegate().setHint( hintName, value );
+		return this;
+	}
+
+	@Override
+	public List<R> list() {
+		return getDelegate().list();
+	}
+
+	@Override
+	public List<R> getResultList() {
+		return getDelegate().getResultList();
+	}
+
+	@Override
+	public ScrollableResults<R> scroll() {
+		return getDelegate().scroll();
+	}
+
+	@Override
+	public ScrollableResults<R> scroll(ScrollMode scrollMode) {
+		return getDelegate().scroll( scrollMode );
+	}
+
+	@Override
+	public Stream<R> getResultStream() {
+		return getDelegate().getResultStream();
+	}
+
+	@Override
+	public Stream<R> stream() {
+		return getDelegate().stream();
+	}
+
+	@Override
+	public R uniqueResult() {
+		return getDelegate().uniqueResult();
+	}
+
+	@Override
+	public R getSingleResult() {
+		return getDelegate().getSingleResult();
+	}
+
+	@Override
+	public R getSingleResultOrNull() {
+		return getDelegate().getSingleResultOrNull();
+	}
+
+	@Override
+	public Optional<R> uniqueResultOptional() {
+		return getDelegate().uniqueResultOptional();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setEntityGraph(EntityGraph<R> graph, GraphSemantic semantic) {
+		getDelegate().setEntityGraph( graph, semantic );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> enableFetchProfile(String profileName) {
+		getDelegate().enableFetchProfile( profileName );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> disableFetchProfile(String profileName) {
+		getDelegate().disableFetchProfile( profileName );
+		return this;
+	}
+
+	@Override
+	public Integer getFetchSize() {
+		return getDelegate().getFetchSize();
+	}
+
+	@Override
+	public boolean isReadOnly() {
+		return getDelegate().isReadOnly();
+	}
+
+	@Override
+	public int getMaxResults() {
+		return getDelegate().getMaxResults();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setMaxResults(int maxResult) {
+		getDelegate().setMaxResults( maxResult );
+		return this;
+	}
+
+	@Override
+	public int getFirstResult() {
+		return getDelegate().getFirstResult();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setFirstResult(int startPosition) {
+		getDelegate().setFirstResult( startPosition );
+		return this;
+	}
+
+	@Override
+	@Incubating
+	public SqmSelectionQueryImplementor<R> setPage(Page page) {
+		getDelegate().setPage( page );
+		return this;
+	}
+
+	@Override
+	public CacheMode getCacheMode() {
+		return getDelegate().getCacheMode();
+	}
+
+	@Override
+	public CacheStoreMode getCacheStoreMode() {
+		return getDelegate().getCacheStoreMode();
+	}
+
+	@Override
+	public CacheRetrieveMode getCacheRetrieveMode() {
+		return getDelegate().getCacheRetrieveMode();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setCacheStoreMode(CacheStoreMode cacheStoreMode) {
+		getDelegate().setCacheStoreMode( cacheStoreMode );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setCacheRetrieveMode(CacheRetrieveMode cacheRetrieveMode) {
+		getDelegate().setCacheRetrieveMode( cacheRetrieveMode );
+		return this;
+	}
+
+	@Override
+	public boolean isCacheable() {
+		return getDelegate().isCacheable();
+	}
+
+	@Override
+	public boolean isQueryPlanCacheable() {
+		return getDelegate().isQueryPlanCacheable();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setQueryPlanCacheable(boolean queryPlanCacheable) {
+		getDelegate().setQueryPlanCacheable( queryPlanCacheable );
+		return this;
+	}
+
+	@Override
+	public String getCacheRegion() {
+		return getDelegate().getCacheRegion();
+	}
+
+	@Override
+	public LockOptions getLockOptions() {
+		return getDelegate().getLockOptions();
+	}
+
+	@Override
+	public LockModeType getLockMode() {
+		return getDelegate().getLockMode();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setLockMode(LockModeType lockMode) {
+		getDelegate().setLockMode( lockMode );
+		return this;
+	}
+
+	@Override
+	public LockMode getHibernateLockMode() {
+		return getDelegate().getHibernateLockMode();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setHibernateLockMode(LockMode lockMode) {
+		getDelegate().setHibernateLockMode( lockMode );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setLockMode(String alias, LockMode lockMode) {
+		getDelegate().setLockMode( alias, lockMode );
+		return this;
+	}
+
+	@Override
+	@Incubating
+	public SqmSelectionQueryImplementor<R> setOrder(List<Order<? super R>> orders) {
+		getDelegate().setOrder( orders );
+		return this;
+	}
+
+	@Override
+	@Incubating
+	public SqmSelectionQueryImplementor<R> setOrder(Order<? super R> order) {
+		getDelegate().setOrder( order );
+		return this;
+	}
+
+	@Override
+	@Remove
+	@Deprecated(since = "6.2")
+	public SqmSelectionQueryImplementor<R> setAliasSpecificLockMode(String alias, LockMode lockMode) {
+		getDelegate().setAliasSpecificLockMode( alias, lockMode );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setFollowOnLocking(boolean enable) {
+		getDelegate().setFollowOnLocking( enable );
+		return this;
+	}
+
+	@Override
+	public String getQueryString() {
+		return getDelegate().getQueryString();
+	}
+
+	@Override
+	public SqmStatement getSqmStatement() {
+		return getDelegate().getSqmStatement();
+	}
+
+	@Override
+	public ParameterMetadata getParameterMetadata() {
+		return getDelegate().getParameterMetadata();
+	}
+
+	@Override
+	public QueryOptions getQueryOptions() {
+		return getDelegate().getQueryOptions();
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(String name, Object value) {
+		getDelegate().setParameter( name, value );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameter(String name, P value, Class<P> type) {
+		getDelegate().setParameter( name, value, type );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameter(String name, P value, BindableType<P> type) {
+		getDelegate().setParameter( name, value, type );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(String name, Instant value, TemporalType temporalType) {
+		getDelegate().setParameter( name, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(String name, Calendar value, TemporalType temporalType) {
+		getDelegate().setParameter( name, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(String name, Date value, TemporalType temporalType) {
+		getDelegate().setParameter( name, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(int position, Object value) {
+		getDelegate().setParameter( position, value );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameter(int position, P value, Class<P> type) {
+		getDelegate().setParameter( position, value, type );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameter(int position, P value, BindableType<P> type) {
+		getDelegate().setParameter( position, value, type );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(int position, Instant value, TemporalType temporalType) {
+		getDelegate().setParameter( position, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(int position, Date value, TemporalType temporalType) {
+		getDelegate().setParameter( position, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(int position, Calendar value, TemporalType temporalType) {
+		getDelegate().setParameter( position, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public <T> SqmSelectionQueryImplementor<R> setParameter(QueryParameter<T> parameter, T value) {
+		getDelegate().setParameter( parameter, value );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameter(QueryParameter<P> parameter, P value, Class<P> type) {
+		getDelegate().setParameter( parameter, value, type );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameter(QueryParameter<P> parameter, P val, BindableType<P> type) {
+		getDelegate().setParameter( parameter, val, type );
+		return this;
+	}
+
+	@Override
+	public <T> SqmSelectionQueryImplementor<R> setParameter(Parameter<T> param, T value) {
+		getDelegate().setParameter( param, value );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(Parameter<Calendar> param, Calendar value, TemporalType temporalType) {
+		getDelegate().setParameter( param, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameter(Parameter<Date> param, Date value, TemporalType temporalType) {
+		getDelegate().setParameter( param, value, temporalType );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameterList(String name, Collection values) {
+		getDelegate().setParameterList( name, values );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(String name, Collection<? extends P> values, Class<P> javaType) {
+		getDelegate().setParameterList( name, values, javaType );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(
+			String name,
+			Collection<? extends P> values,
+			BindableType<P> type) {
+		getDelegate().setParameterList( name, values, type );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameterList(String name, Object[] values) {
+		getDelegate().setParameterList( name, values );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(String name, P[] values, Class<P> javaType) {
+		getDelegate().setParameterList( name, values, javaType );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(String name, P[] values, BindableType<P> type) {
+		getDelegate().setParameterList( name, values, type );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameterList(int position, Collection values) {
+		getDelegate().setParameterList( position, values );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(int position, Collection<? extends P> values, Class<P> javaType) {
+		getDelegate().setParameterList( position, values, javaType );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(
+			int position,
+			Collection<? extends P> values,
+			BindableType<P> type) {
+		getDelegate().setParameterList( position, values, type );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setParameterList(int position, Object[] values) {
+		getDelegate().setParameterList( position, values );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(int position, P[] values, Class<P> javaType) {
+		getDelegate().setParameterList( position, values, javaType );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(int position, P[] values, BindableType<P> type) {
+		getDelegate().setParameterList( position, values, type );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(QueryParameter<P> parameter, Collection<? extends P> values) {
+		getDelegate().setParameterList( parameter, values );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(
+			QueryParameter<P> parameter,
+			Collection<? extends P> values,
+			Class<P> javaType) {
+		getDelegate().setParameterList( parameter, values, javaType );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(
+			QueryParameter<P> parameter,
+			Collection<? extends P> values,
+			BindableType<P> type) {
+		getDelegate().setParameterList( parameter, values, type );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(QueryParameter<P> parameter, P[] values) {
+		getDelegate().setParameterList( parameter, values );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(QueryParameter<P> parameter, P[] values, Class<P> javaType) {
+		getDelegate().setParameterList( parameter, values, javaType );
+		return this;
+	}
+
+	@Override
+	public <P> SqmSelectionQueryImplementor<R> setParameterList(QueryParameter<P> parameter, P[] values, BindableType<P> type) {
+		getDelegate().setParameterList( parameter, values, type );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setProperties(Object bean) {
+		getDelegate().setProperties( bean );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setProperties(Map bean) {
+		getDelegate().setProperties( bean );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setHibernateFlushMode(FlushMode flushMode) {
+		getDelegate().setHibernateFlushMode( flushMode );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setCacheMode(CacheMode cacheMode) {
+		getDelegate().setCacheMode( cacheMode );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setCacheable(boolean cacheable) {
+		getDelegate().setCacheable( cacheable );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setCacheRegion(String cacheRegion) {
+		getDelegate().setCacheRegion( cacheRegion );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setTimeout(int timeout) {
+		getDelegate().setTimeout( timeout );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setFetchSize(int fetchSize) {
+		getDelegate().setFetchSize( fetchSize );
+		return this;
+	}
+
+	@Override
+	public SqmSelectionQueryImplementor<R> setReadOnly(boolean readOnly) {
+		getDelegate().setReadOnly( readOnly );
+		return this;
+	}
+
+	@Override
+	public <T> T executeQuery(ResultsConsumer<T, R> resultsConsumer) {
+		return getDelegate().executeQuery( resultsConsumer );
+	}
+}

hibernate-core/src/main/java/org/hibernate/query/sqm/spi/SqmSelectionQueryImplementor.java

@@ -0,0 +1,19 @@
+/*
+ * 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.sqm.spi;
+
+import org.hibernate.Incubating;
+import org.hibernate.query.sqm.SqmSelectionQuery;
+import org.hibernate.sql.results.spi.ResultsConsumer;
+
+/**
+ * @since 6.4
+ */
+@Incubating
+public interface SqmSelectionQueryImplementor<R> extends SqmSelectionQuery<R> {
+	<T> T executeQuery(ResultsConsumer<T, R> resultsConsumer);
+}

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

@@ -10,33 +10,19 @@
 import java.sql.PreparedStatement;
 import java.util.List;
 import java.util.Set;
-import java.util.Spliterator;
-import java.util.Spliterators;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
-import java.util.stream.Stream;
-import java.util.stream.StreamSupport;
 
 import org.hibernate.CacheMode;
-import org.hibernate.FlushMode;
 import org.hibernate.LockOptions;
-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.graph.spi.AppliedGraph;
 import org.hibernate.internal.util.collections.ArrayHelper;
-import org.hibernate.query.ResultListTransformer;
 import org.hibernate.query.TupleTransformer;
-import org.hibernate.query.internal.ScrollableResultsIterator;
-import org.hibernate.query.spi.Limit;
-import org.hibernate.query.spi.QueryOptions;
-import org.hibernate.query.spi.QueryParameterBindings;
-import org.hibernate.query.spi.ScrollableResultsImplementor;
 import org.hibernate.sql.exec.SqlExecLogger;
-import org.hibernate.sql.exec.spi.Callback;
 import org.hibernate.sql.exec.spi.ExecutionContext;
 import org.hibernate.sql.exec.spi.JdbcOperationQuerySelect;
 import org.hibernate.sql.exec.spi.JdbcParameterBindings;
@@ -56,19 +42,14 @@
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesMappingProducer;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesMetadata;
 import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
-import org.hibernate.sql.results.spi.ListResultsConsumer;
 import org.hibernate.sql.results.spi.ResultsConsumer;
 import org.hibernate.sql.results.spi.RowReader;
 import org.hibernate.sql.results.spi.RowTransformer;
-import org.hibernate.sql.results.spi.ScrollableResultsConsumer;
 import org.hibernate.stat.spi.StatisticsImplementor;
 import org.hibernate.type.BasicType;
 import org.hibernate.type.descriptor.java.JavaType;
 import org.hibernate.type.spi.TypeConfiguration;
 
-import jakarta.persistence.CacheRetrieveMode;
-import jakarta.persistence.CacheStoreMode;
-
 /**
  * Standard JdbcSelectExecutor implementation used by Hibernate,
  * through {@link JdbcSelectExecutorStandardImpl#INSTANCE}
@@ -82,73 +63,7 @@ public class JdbcSelectExecutorStandardImpl implements JdbcSelectExecutor {
 	public static final JdbcSelectExecutorStandardImpl INSTANCE = new JdbcSelectExecutorStandardImpl();
 
 	@Override
-	public <R> List<R> list(
+	public <T, R> T executeQuery(
-			JdbcOperationQuerySelect jdbcSelect,
-			JdbcParameterBindings jdbcParameterBindings,
-			ExecutionContext executionContext,
-			RowTransformer<R> rowTransformer,
-			Class<R> domainResultType,
-			ListResultsConsumer.UniqueSemantic uniqueSemantic) {
-		// Only do auto flushing for top level queries
-		return executeQuery(
-				jdbcSelect,
-				jdbcParameterBindings,
-				executionContext,
-				rowTransformer,
-				domainResultType,
-				(sql) -> executionContext.getSession()
-						.getJdbcCoordinator()
-						.getStatementPreparer()
-						.prepareQueryStatement( sql, false, null ),
-				ListResultsConsumer.instance( uniqueSemantic )
-		);
-	}
-
-	@Override
-	public <R> ScrollableResultsImplementor<R> scroll(
-			JdbcOperationQuerySelect jdbcSelect,
-			ScrollMode scrollMode,
-			JdbcParameterBindings jdbcParameterBindings,
-			ExecutionContext executionContext,
-			RowTransformer<R> rowTransformer) {
-		final SharedSessionContractImplementor session = executionContext.getSession();
-		session.autoFlushIfRequired( jdbcSelect.getAffectedTableNames() );
-		return executeQueryScroll(
-				jdbcSelect,
-				jdbcParameterBindings,
-				executionContext,
-				rowTransformer,
-				null,
-				(sql) -> executionContext.getSession().getJdbcCoordinator().getStatementPreparer().prepareQueryStatement(
-						sql,
-						false,
-						scrollMode
-				),
-				ScrollableResultsConsumer.instance()
-		);
-	}
-
-	@Override
-	public <R> Stream<R> stream(
-			JdbcOperationQuerySelect jdbcSelect,
-			JdbcParameterBindings jdbcParameterBindings,
-			ExecutionContext executionContext,
-			RowTransformer<R> rowTransformer) {
-		final ScrollableResultsImplementor<R> scrollableResults = scroll(
-				jdbcSelect,
-				ScrollMode.FORWARD_ONLY,
-				jdbcParameterBindings,
-				executionContext,
-				rowTransformer
-		);
-		final ScrollableResultsIterator<R> iterator = new ScrollableResultsIterator<>( scrollableResults );
-		final Spliterator<R> spliterator = Spliterators.spliteratorUnknownSize( iterator, Spliterator.NONNULL );
-
-		final Stream<R> stream = StreamSupport.stream( spliterator, false );
-		return stream.onClose( scrollableResults::close );
-	}
-
-	private <T, R> T executeQuery(
 			JdbcOperationQuerySelect jdbcSelect,
 			JdbcParameterBindings jdbcParameterBindings,
 			ExecutionContext executionContext,
@@ -182,73 +97,6 @@ private <T, R> T executeQuery(
 		}
 	}
 
-	private <T, R> T executeQueryScroll(
-			JdbcOperationQuerySelect jdbcSelect,
-			JdbcParameterBindings jdbcParameterBindings,
-			ExecutionContext executionContext,
-			RowTransformer<R> rowTransformer,
-			Class<R> domainResultType,
-			Function<String, PreparedStatement> statementCreator,
-			ResultsConsumer<T, R> resultsConsumer) {
-		return doExecuteQuery(
-				jdbcSelect,
-				jdbcParameterBindings,
-				getScrollContext( executionContext, executionContext.getSession().getPersistenceContext() ),
-				rowTransformer,
-				domainResultType,
-				statementCreator,
-				resultsConsumer
-		);
-	}
-
-	/*
-		When `Query#scroll()` is call the query is not executed immediately, a new ExecutionContext with the values of the `persistenceContext.isDefaultReadOnly()` and of the `queryOptions.isReadOnly()`
-		set at the moment of the Query#scroll() call is created in order to use it when the query will be executed.
-	 */
-	private ExecutionContext getScrollContext(ExecutionContext context, PersistenceContext persistenceContext) {
-		final QueryOptions queryOptions = context.getQueryOptions();
-		final Boolean readOnly;
-		if ( queryOptions.isReadOnly() == null ) {
-			readOnly = persistenceContext.isDefaultReadOnly();
-		}
-		else {
-			readOnly = queryOptions.isReadOnly();
-		}
-		final Integer timeout = queryOptions.getTimeout();
-		final FlushMode flushMode = queryOptions.getFlushMode();
-		final AppliedGraph appliedGraph = queryOptions.getAppliedGraph();
-		final TupleTransformer<?> tupleTransformer = queryOptions.getTupleTransformer();
-		final ResultListTransformer<?> resultListTransformer = queryOptions.getResultListTransformer();
-		final Boolean resultCachingEnabled = queryOptions.isResultCachingEnabled();
-		final CacheRetrieveMode cacheRetrieveMode = queryOptions.getCacheRetrieveMode();
-		final CacheStoreMode cacheStoreMode = queryOptions.getCacheStoreMode();
-		final String resultCacheRegionName = queryOptions.getResultCacheRegionName();
-		final LockOptions lockOptions = queryOptions.getLockOptions();
-		final String comment = queryOptions.getComment();
-		final List<String> databaseHints = queryOptions.getDatabaseHints();
-		final Integer fetchSize = queryOptions.getFetchSize();
-		final Limit limit = queryOptions.getLimit();
-
-		return new JdbcSelectExecutionContext(
-				timeout,
-				flushMode,
-				readOnly,
-				appliedGraph,
-				tupleTransformer,
-				resultListTransformer,
-				resultCachingEnabled,
-				cacheRetrieveMode,
-				cacheStoreMode,
-				resultCacheRegionName,
-				lockOptions,
-				comment,
-				databaseHints,
-				fetchSize,
-				limit,
-				context
-		);
-	}
-
 	private <T, R> T doExecuteQuery(
 			JdbcOperationQuerySelect jdbcSelect,
 			JdbcParameterBindings jdbcParameterBindings,
@@ -642,175 +490,4 @@ public <J> BasicType<J> resolveType(
 		}
 
 	}
-
-	private static class JdbcSelectExecutionContext extends BaseExecutionContext implements QueryOptions {
-
-		private final Integer timeout;
-		private final FlushMode flushMode;
-		private final Boolean readOnly;
-		private final AppliedGraph appliedGraph;
-		private final TupleTransformer<?> tupleTransformer;
-		private final ResultListTransformer<?> resultListTransformer;
-		private final Boolean resultCachingEnabled;
-		private final CacheRetrieveMode cacheRetrieveMode;
-		private final CacheStoreMode cacheStoreMode;
-		private final String resultCacheRegionName;
-		private final LockOptions lockOptions;
-		private final String comment;
-		private final List<String> databaseHints;
-		private final Integer fetchSize;
-		private final Limit limit;
-		private final ExecutionContext context;
-
-		public JdbcSelectExecutionContext(
-				Integer timeout,
-				FlushMode flushMode,
-				Boolean readOnly,
-				AppliedGraph appliedGraph,
-				TupleTransformer<?> tupleTransformer,
-				ResultListTransformer<?> resultListTransformer,
-				Boolean resultCachingEnabled,
-				CacheRetrieveMode cacheRetrieveMode,
-				CacheStoreMode cacheStoreMode,
-				String resultCacheRegionName,
-				LockOptions lockOptions,
-				String comment,
-				List<String> databaseHints,
-				Integer fetchSize,
-				Limit limit,
-				ExecutionContext context) {
-			super( context.getSession() );
-			this.timeout = timeout;
-			this.flushMode = flushMode;
-			this.readOnly = readOnly;
-			this.appliedGraph = appliedGraph;
-			this.tupleTransformer = tupleTransformer;
-			this.resultListTransformer = resultListTransformer;
-			this.resultCachingEnabled = resultCachingEnabled;
-			this.cacheRetrieveMode = cacheRetrieveMode;
-			this.cacheStoreMode = cacheStoreMode;
-			this.resultCacheRegionName = resultCacheRegionName;
-			this.lockOptions = lockOptions;
-			this.comment = comment;
-			this.databaseHints = databaseHints;
-			this.fetchSize = fetchSize;
-			this.limit = limit;
-			this.context = context;
-		}
-
-		@Override
-		public boolean isScrollResult() {
-			return true;
-		}
-
-		@Override
-		public QueryOptions getQueryOptions() {
-			return this;
-		}
-
-		@Override
-		public Integer getTimeout() {
-			return timeout;
-		}
-
-		@Override
-		public FlushMode getFlushMode() {
-			return flushMode;
-		}
-
-		@Override
-		public Boolean isReadOnly() {
-			return readOnly;
-		}
-
-		@Override
-		public AppliedGraph getAppliedGraph() {
-			return appliedGraph;
-		}
-
-		@Override
-		public TupleTransformer<?> getTupleTransformer() {
-			return tupleTransformer;
-		}
-
-		@Override
-		public ResultListTransformer<?> getResultListTransformer() {
-			return resultListTransformer;
-		}
-
-		@Override
-		public Boolean isResultCachingEnabled() {
-			return resultCachingEnabled;
-		}
-
-		@Override
-		public Boolean getQueryPlanCachingEnabled() {
-			return null;
-		}
-
-		@Override
-		public CacheRetrieveMode getCacheRetrieveMode() {
-			return cacheRetrieveMode;
-		}
-
-		@Override
-		public CacheStoreMode getCacheStoreMode() {
-			return cacheStoreMode;
-		}
-
-		@Override
-		public String getResultCacheRegionName() {
-			return resultCacheRegionName;
-		}
-
-		@Override
-		public LockOptions getLockOptions() {
-			return lockOptions;
-		}
-
-		@Override
-		public String getComment() {
-			return comment;
-		}
-
-		@Override
-		public List<String> getDatabaseHints() {
-			return databaseHints;
-		}
-
-		@Override
-		public Integer getFetchSize() {
-			return fetchSize;
-		}
-
-		@Override
-		public Limit getLimit() {
-			return limit;
-		}
-
-		@Override
-		public QueryParameterBindings getQueryParameterBindings() {
-			return context.getQueryParameterBindings();
-		}
-
-		@Override
-		public Callback getCallback() {
-			return context.getCallback();
-		}
-
-		@Override
-		public boolean hasCallbackActions() {
-			return context.hasCallbackActions();
-		}
-
-		@Override
-		public Set<String> getEnabledFetchProfiles() {
-			return null;
-		}
-
-		@Override
-		public Set<String> getDisabledFetchProfiles() {
-			return null;
-		}
-	}
 }

hibernate-core/src/main/java/org/hibernate/sql/exec/spi/JdbcSelectExecutor.java

@@ -6,14 +6,35 @@
  */
 package org.hibernate.sql.exec.spi;
 
+import java.sql.PreparedStatement;
 import java.util.List;
+import java.util.Set;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.function.Function;
 import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
 
+import org.hibernate.FlushMode;
 import org.hibernate.Incubating;
+import org.hibernate.LockOptions;
 import org.hibernate.ScrollMode;
+import org.hibernate.graph.spi.AppliedGraph;
+import org.hibernate.query.ResultListTransformer;
+import org.hibernate.query.TupleTransformer;
+import org.hibernate.query.internal.ScrollableResultsIterator;
+import org.hibernate.query.spi.Limit;
+import org.hibernate.query.spi.QueryOptions;
+import org.hibernate.query.spi.QueryParameterBindings;
 import org.hibernate.query.spi.ScrollableResultsImplementor;
+import org.hibernate.sql.exec.internal.BaseExecutionContext;
 import org.hibernate.sql.results.spi.ListResultsConsumer;
+import org.hibernate.sql.results.spi.ResultsConsumer;
 import org.hibernate.sql.results.spi.RowTransformer;
+import org.hibernate.sql.results.spi.ScrollableResultsConsumer;
+
+import jakarta.persistence.CacheRetrieveMode;
+import jakarta.persistence.CacheStoreMode;
 
 /**
  * An executor for JdbcSelect operations.
@@ -22,6 +43,19 @@
  */
 @Incubating
 public interface JdbcSelectExecutor {
+
+	/**
+	 * @since 6.4
+	 */
+	<T, R> T executeQuery(
+			JdbcOperationQuerySelect jdbcSelect,
+			JdbcParameterBindings jdbcParameterBindings,
+			ExecutionContext executionContext,
+			RowTransformer<R> rowTransformer,
+			Class<R> domainResultType,
+			Function<String, PreparedStatement> statementCreator,
+			ResultsConsumer<T, R> resultsConsumer);
+
 	default <R> List<R> list(
 			JdbcOperationQuerySelect jdbcSelect,
 			JdbcParameterBindings jdbcParameterBindings,
@@ -31,24 +65,267 @@ default <R> List<R> list(
 		return list( jdbcSelect, jdbcParameterBindings, executionContext, rowTransformer, null, uniqueSemantic );
 	}
 
-	<R> List<R> list(
+	default <R> List<R> list(
 			JdbcOperationQuerySelect jdbcSelect,
 			JdbcParameterBindings jdbcParameterBindings,
 			ExecutionContext executionContext,
 			RowTransformer<R> rowTransformer,
 			Class<R> requestedJavaType,
-			ListResultsConsumer.UniqueSemantic uniqueSemantic);
+			ListResultsConsumer.UniqueSemantic uniqueSemantic) {
+		// Only do auto flushing for top level queries
+		return executeQuery(
+				jdbcSelect,
+				jdbcParameterBindings,
+				executionContext,
+				rowTransformer,
+				requestedJavaType,
+				sql -> executionContext.getSession()
+						.getJdbcCoordinator()
+						.getStatementPreparer()
+						.prepareQueryStatement( sql, false, null ),
+				ListResultsConsumer.instance( uniqueSemantic )
+		);
+	}
 
-	<R> ScrollableResultsImplementor<R> scroll(
+	default <R> ScrollableResultsImplementor<R> scroll(
 			JdbcOperationQuerySelect jdbcSelect,
 			ScrollMode scrollMode,
 			JdbcParameterBindings jdbcParameterBindings,
 			ExecutionContext executionContext,
-			RowTransformer<R> rowTransformer);
+			RowTransformer<R> rowTransformer) {
+		return executeQuery(
+				jdbcSelect,
+				jdbcParameterBindings,
+				getScrollContext( executionContext ),
+				rowTransformer,
+				null,
+				sql -> executionContext.getSession().getJdbcCoordinator().getStatementPreparer().prepareQueryStatement(
+						sql,
+						false,
+						scrollMode
+				),
+				ScrollableResultsConsumer.instance()
+		);
+	}
+
+
+	/*
+		When `Query#scroll()` is call the query is not executed immediately, a new ExecutionContext with the values of the `persistenceContext.isDefaultReadOnly()` and of the `queryOptions.isReadOnly()`
+		set at the moment of the Query#scroll() call is created in order to use it when the query will be executed.
+	 */
+	private ExecutionContext getScrollContext(ExecutionContext context) {
+		class ScrollableExecutionContext extends BaseExecutionContext implements QueryOptions {
+
+			private final Integer timeout;
+			private final FlushMode flushMode;
+			private final Boolean readOnly;
+			private final AppliedGraph appliedGraph;
+			private final TupleTransformer<?> tupleTransformer;
+			private final ResultListTransformer<?> resultListTransformer;
+			private final Boolean resultCachingEnabled;
+			private final CacheRetrieveMode cacheRetrieveMode;
+			private final CacheStoreMode cacheStoreMode;
+			private final String resultCacheRegionName;
+			private final LockOptions lockOptions;
+			private final String comment;
+			private final List<String> databaseHints;
+			private final Integer fetchSize;
+			private final Limit limit;
+			private final ExecutionContext context;
+
+			public ScrollableExecutionContext(
+					Integer timeout,
+					FlushMode flushMode,
+					Boolean readOnly,
+					AppliedGraph appliedGraph,
+					TupleTransformer<?> tupleTransformer,
+					ResultListTransformer<?> resultListTransformer,
+					Boolean resultCachingEnabled,
+					CacheRetrieveMode cacheRetrieveMode,
+					CacheStoreMode cacheStoreMode,
+					String resultCacheRegionName,
+					LockOptions lockOptions,
+					String comment,
+					List<String> databaseHints,
+					Integer fetchSize,
+					Limit limit,
+					ExecutionContext context) {
+				super( context.getSession() );
+				this.timeout = timeout;
+				this.flushMode = flushMode;
+				this.readOnly = readOnly;
+				this.appliedGraph = appliedGraph;
+				this.tupleTransformer = tupleTransformer;
+				this.resultListTransformer = resultListTransformer;
+				this.resultCachingEnabled = resultCachingEnabled;
+				this.cacheRetrieveMode = cacheRetrieveMode;
+				this.cacheStoreMode = cacheStoreMode;
+				this.resultCacheRegionName = resultCacheRegionName;
+				this.lockOptions = lockOptions;
+				this.comment = comment;
+				this.databaseHints = databaseHints;
+				this.fetchSize = fetchSize;
+				this.limit = limit;
+				this.context = context;
+			}
+
+			@Override
+			public boolean isScrollResult() {
+				return true;
+			}
+
+			@Override
+			public QueryOptions getQueryOptions() {
+				return this;
+			}
+
+			@Override
+			public Integer getTimeout() {
+				return timeout;
+			}
+
+			@Override
+			public FlushMode getFlushMode() {
+				return flushMode;
+			}
+
+			@Override
+			public Boolean isReadOnly() {
+				return readOnly;
+			}
+
+			@Override
+			public AppliedGraph getAppliedGraph() {
+				return appliedGraph;
+			}
+
+			@Override
+			public TupleTransformer<?> getTupleTransformer() {
+				return tupleTransformer;
+			}
+
+			@Override
+			public ResultListTransformer<?> getResultListTransformer() {
+				return resultListTransformer;
+			}
+
+			@Override
+			public Boolean isResultCachingEnabled() {
+				return resultCachingEnabled;
+			}
+
+			@Override
+			public Boolean getQueryPlanCachingEnabled() {
+				return null;
+			}
+
+			@Override
+			public CacheRetrieveMode getCacheRetrieveMode() {
+				return cacheRetrieveMode;
+			}
+
+			@Override
+			public CacheStoreMode getCacheStoreMode() {
+				return cacheStoreMode;
+			}
+
+			@Override
+			public String getResultCacheRegionName() {
+				return resultCacheRegionName;
+			}
+
+			@Override
+			public LockOptions getLockOptions() {
+				return lockOptions;
+			}
+
+			@Override
+			public String getComment() {
+				return comment;
+			}
+
+			@Override
+			public List<String> getDatabaseHints() {
+				return databaseHints;
+			}
+
+			@Override
+			public Integer getFetchSize() {
+				return fetchSize;
+			}
+
+			@Override
+			public Limit getLimit() {
+				return limit;
+			}
+
+			@Override
+			public QueryParameterBindings getQueryParameterBindings() {
+				return context.getQueryParameterBindings();
+			}
+
+			@Override
+			public Callback getCallback() {
+				return context.getCallback();
+			}
+
+			@Override
+			public boolean hasCallbackActions() {
+				return context.hasCallbackActions();
+			}
+
+			@Override
+			public Set<String> getEnabledFetchProfiles() {
+				return null;
+			}
+
+			@Override
+			public Set<String> getDisabledFetchProfiles() {
+				return null;
+			}
+		}
+
+		final QueryOptions queryOptions = context.getQueryOptions();
+		final Boolean readOnly;
+		if ( queryOptions.isReadOnly() == null ) {
+			readOnly = context.getSession().getPersistenceContext().isDefaultReadOnly();
+		}
+		else {
+			readOnly = queryOptions.isReadOnly();
+		}
+		final Integer timeout = queryOptions.getTimeout();
+		final FlushMode flushMode = queryOptions.getFlushMode();
+		final AppliedGraph appliedGraph = queryOptions.getAppliedGraph();
+		final TupleTransformer<?> tupleTransformer = queryOptions.getTupleTransformer();
+		final ResultListTransformer<?> resultListTransformer = queryOptions.getResultListTransformer();
+		final Boolean resultCachingEnabled = queryOptions.isResultCachingEnabled();
+		final CacheRetrieveMode cacheRetrieveMode = queryOptions.getCacheRetrieveMode();
+		final CacheStoreMode cacheStoreMode = queryOptions.getCacheStoreMode();
+		final String resultCacheRegionName = queryOptions.getResultCacheRegionName();
+		final LockOptions lockOptions = queryOptions.getLockOptions();
+		final String comment = queryOptions.getComment();
+		final List<String> databaseHints = queryOptions.getDatabaseHints();
+		final Integer fetchSize = queryOptions.getFetchSize();
+		final Limit limit = queryOptions.getLimit();
+
+		return new ScrollableExecutionContext(
+				timeout,
+				flushMode,
+				readOnly,
+				appliedGraph,
+				tupleTransformer,
+				resultListTransformer,
+				resultCachingEnabled,
+				cacheRetrieveMode,
+				cacheStoreMode,
+				resultCacheRegionName,
+				lockOptions,
+				comment,
+				databaseHints,
+				fetchSize,
+				limit,
+				context
+		);
+	}
 
-	<R> Stream<R> stream(
-			JdbcOperationQuerySelect jdbcSelect,
-			JdbcParameterBindings jdbcParameterBindings,
-			ExecutionContext executionContext,
-			RowTransformer<R> rowTransformer);
 }

hibernate-core/src/test/java/org/hibernate/orm/test/query/sqm/DelegatingSqmSelectionQueryImplementorTest.java

@@ -0,0 +1,14 @@
+package org.hibernate.orm.test.query.sqm;
+
+import org.hibernate.query.sqm.spi.DelegatingSqmSelectionQueryImplementor;
+import org.hibernate.query.sqm.spi.SqmSelectionQueryImplementor;
+
+/**
+ * This class just serves as compilation unit to verify we implemented all methods in the {@link DelegatingSqmSelectionQueryImplementor} class.
+ */
+public class DelegatingSqmSelectionQueryImplementorTest<R> extends DelegatingSqmSelectionQueryImplementor<R> {
+	@Override
+	protected SqmSelectionQueryImplementor<R> getDelegate() {
+		return null;
+	}
+}