Enable Single Query Loading for simple aggregates.

Single Query Loading loads as the name suggests complete aggregates using a single select.
While the ultimate goal is to support this for all aggregates, this commit enables it only for simple aggregate and also only for some operations.
A simple aggregate is an aggregate that only reference up to one other entity and does not have embedded entities.
The supported operations are those available via `CrudRepository`: `findAll`, `findById`, and `findAllByIds`.

Single Query Loading does NOT work with the supported in memory databases H2 and HSQLDB, since these do not properly support windowing functions, which are essential for Single Query Loading.

To turn on Single Query Loading call `RelationalMappingContext.setSingleQueryLoadingEnabled(true)`.

Closes #1446
See #1450
See #1445
Original pull request: #1572

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

@@ -0,0 +1,134 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.jdbc.core.convert;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.relational.core.dialect.Dialect;
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.sqlgeneration.AliasFactory;
+import org.springframework.data.relational.core.sqlgeneration.CachingSqlGenerator;
+import org.springframework.data.relational.core.sqlgeneration.SingleQuerySqlGenerator;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+import org.springframework.util.Assert;
+
+/**
+ * Reads complete Aggregates from the database, by generating appropriate SQL using a {@link SingleQuerySqlGenerator}
+ * and a matching {@link AggregateResultSetExtractor} and invoking a
+ * {@link org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate}
+ * 
+ * @param <T> the type of aggregate produced by this reader.
+ * @since 3.2
+ * @author Jens Schauder
+ */
+class AggregateReader<T> {
+
+	private final RelationalMappingContext mappingContext;
+	private final RelationalPersistentEntity<T> aggregate;
+	private final AliasFactory aliasFactory;
+	private final org.springframework.data.relational.core.sqlgeneration.SqlGenerator sqlGenerator;
+	private final JdbcConverter converter;
+	private final NamedParameterJdbcOperations jdbcTemplate;
+
+	AggregateReader(RelationalMappingContext mappingContext, Dialect dialect, JdbcConverter converter,
+			NamedParameterJdbcOperations jdbcTemplate, RelationalPersistentEntity<T> aggregate) {
+
+		this.mappingContext = mappingContext;
+
+		this.aggregate = aggregate;
+		this.converter = converter;
+		this.jdbcTemplate = jdbcTemplate;
+
+		this.sqlGenerator = new CachingSqlGenerator(new SingleQuerySqlGenerator(mappingContext, dialect, aggregate));
+		this.aliasFactory = sqlGenerator.getAliasFactory();
+	}
+
+	public List<T> findAll() {
+
+		String sql = sqlGenerator.findAll();
+
+		PathToColumnMapping pathToColumn = createPathToColumnMapping(aliasFactory);
+		AggregateResultSetExtractor<T> extractor = new AggregateResultSetExtractor<>(mappingContext, aggregate, converter,
+				pathToColumn);
+
+		Iterable<T> result = jdbcTemplate.query(sql, extractor);
+
+		Assert.state(result != null, "result is null");
+
+		return (List<T>) result;
+	}
+
+	public T findById(Object id) {
+
+		PathToColumnMapping pathToColumn = createPathToColumnMapping(aliasFactory);
+		AggregateResultSetExtractor<T> extractor = new AggregateResultSetExtractor<>(mappingContext, aggregate, converter,
+				pathToColumn);
+
+		String sql = sqlGenerator.findById();
+
+		id = converter.writeValue(id, aggregate.getRequiredIdProperty().getTypeInformation());
+
+		Iterator<T> result = jdbcTemplate.query(sql, Map.of("id", id), extractor).iterator();
+
+		T returnValue = result.hasNext() ? result.next() : null;
+
+		if (result.hasNext()) {
+			throw new IncorrectResultSizeDataAccessException(1);
+		}
+
+		return returnValue;
+	}
+
+	public Iterable<T> findAllById(Iterable<?> ids) {
+
+		PathToColumnMapping pathToColumn = createPathToColumnMapping(aliasFactory);
+		AggregateResultSetExtractor<T> extractor = new AggregateResultSetExtractor<>(mappingContext, aggregate, converter,
+				pathToColumn);
+
+		String sql = sqlGenerator.findAllById();
+
+		List<Object> convertedIds = new ArrayList<>();
+		for (Object id : ids) {
+			convertedIds.add(converter.writeValue(id, aggregate.getRequiredIdProperty().getTypeInformation()));
+		}
+
+		return jdbcTemplate.query(sql, Map.of("ids", convertedIds), extractor);
+	}
+
+	private PathToColumnMapping createPathToColumnMapping(AliasFactory aliasFactory) {
+		return new PathToColumnMapping() {
+			@Override
+			public String column(AggregatePath path) {
+
+				String alias = aliasFactory.getColumnAlias(path);
+				Assert.notNull(alias, () -> "alias for >" + path + "<must not be null");
+				return alias;
+			}
+
+			@Override
+			public String keyColumn(AggregatePath path) {
+				return aliasFactory.getKeyAlias(path);
+			}
+		};
+	}
+}

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.jdbc.core.convert;
+
+import org.springframework.data.relational.core.dialect.Dialect;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+
+/**
+ * Creates {@link AggregateReader} instances.
+ * 
+ * @since 3.2
+ * @author Jens Schauder
+ */
+class AggregateReaderFactory {
+
+	private final RelationalMappingContext mappingContext;
+	private final Dialect dialect;
+	private final JdbcConverter converter;
+	private final NamedParameterJdbcOperations jdbcTemplate;
+
+	public AggregateReaderFactory(RelationalMappingContext mappingContext, Dialect dialect, JdbcConverter converter,
+			NamedParameterJdbcOperations jdbcTemplate) {
+
+		this.mappingContext = mappingContext;
+		this.dialect = dialect;
+		this.converter = converter;
+		this.jdbcTemplate = jdbcTemplate;
+	}
+
+	<T> AggregateReader<T> createAggregateReaderFor(RelationalPersistentEntity<T> entity) {
+		return new AggregateReader<>(mappingContext, dialect, converter, jdbcTemplate, entity);
+	}
+}

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

@@ -0,0 +1,596 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.convert;
+
+import java.sql.ResultSet;
+import java.util.AbstractCollection;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.mapping.Parameter;
+import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PropertyHandler;
+import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.data.mapping.model.EntityInstantiator;
+import org.springframework.data.mapping.model.ParameterValueProvider;
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.jdbc.core.ResultSetExtractor;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Extracts complete aggregates from a {@link ResultSet}. The {@literal ResultSet} must have a very special structure
+ * which looks somewhat how one would represent an aggregate in a single excel table. The first row contains data of the
+ * aggregate root, any single valued reference and the first element of any collection. Following rows do NOT repeat the
+ * aggregate root data but contain data of second elements of any collections. For details see accompanying unit tests.
+ * 
+ * @param <T> the type of aggregates to extract
+ * @since 3.2
+ * @author Jens Schauder
+ */
+class AggregateResultSetExtractor<T> implements ResultSetExtractor<Iterable<T>> {
+
+	private final RelationalMappingContext context;
+	private final RelationalPersistentEntity<T> rootEntity;
+	private final JdbcConverter converter;
+	private final PathToColumnMapping propertyToColumn;
+
+	/**
+	 * @param context the {@link org.springframework.data.mapping.context.MappingContext} providing the metadata for the
+	 *          aggregate and its entity. Must not be {@literal null}.
+	 * @param rootEntity the aggregate root. Must not be {@literal null}.
+	 * @param converter Used for converting objects from the database to whatever is required by the aggregate. Must not
+	 *          be {@literal null}.
+	 * @param pathToColumn a mapping from {@link org.springframework.data.relational.core.mapping.AggregatePath} to the
+	 *          column of the {@link ResultSet} that holds the data for that
+	 *          {@link org.springframework.data.relational.core.mapping.AggregatePath}.
+	 */
+	AggregateResultSetExtractor(RelationalMappingContext context, RelationalPersistentEntity<T> rootEntity,
+			JdbcConverter converter, PathToColumnMapping pathToColumn) {
+
+		Assert.notNull(context, "context must not be null");
+		Assert.notNull(rootEntity, "rootEntity must not be null");
+		Assert.notNull(converter, "converter must not be null");
+		Assert.notNull(pathToColumn, "propertyToColumn must not be null");
+
+		this.context = context;
+		this.rootEntity = rootEntity;
+		this.converter = converter;
+		this.propertyToColumn = pathToColumn;
+	}
+
+	@Override
+	public Iterable<T> extractData(ResultSet resultSet) throws DataAccessException {
+
+		CachingResultSet crs = new CachingResultSet(resultSet);
+
+		CollectionReader reader = new CollectionReader(crs);
+
+		while (crs.next()) {
+			reader.read();
+		}
+
+		return (Iterable<T>) reader.getResultAndReset();
+	}
+
+	/**
+	 * create an instance and populate all its properties
+	 */
+	@Nullable
+	private Object hydrateInstance(EntityInstantiator instantiator, ResultSetParameterValueProvider valueProvider,
+			RelationalPersistentEntity<?> entity) {
+
+		if (!valueProvider.basePath.isRoot() && // this is a nested ValueProvider
+				valueProvider.basePath.getRequiredLeafProperty().isEmbedded() && // it's an embedded
+				!valueProvider.basePath.getRequiredLeafProperty().shouldCreateEmptyEmbedded() && // it's embedded
+				!valueProvider.hasValue()) { // all values have been null
+			return null;
+		}
+
+		Object instance = instantiator.createInstance(entity, valueProvider);
+
+		PersistentPropertyAccessor<?> accessor = new ConvertingPropertyAccessor<>(entity.getPropertyAccessor(instance),
+				converter.getConversionService());
+
+		if (entity.requiresPropertyPopulation()) {
+
+			entity.doWithProperties((PropertyHandler<RelationalPersistentProperty>) p -> {
+
+				if (!entity.isCreatorArgument(p)) {
+					accessor.setProperty(p, valueProvider.getValue(p));
+				}
+			});
+		}
+		return instance;
+	}
+
+	/**
+	 * A {@link Reader} is responsible for reading a single entity or collection of entities from a set of columns
+	 * 
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	private interface Reader {
+
+		/**
+		 * read the data needed for creating the result of this {@literal Reader}
+		 */
+		void read();
+
+		/**
+		 * Checks if this {@literal Reader} has all the data needed for a complete result, or if it needs to read further
+		 * rows.
+		 * 
+		 * @return the result of the check.
+		 */
+		boolean hasResult();
+
+		/**
+		 * Constructs the result, returns it and resets the state of the reader to read the next instance.
+		 * 
+		 * @return an instance of whatever this {@literal Reader} is supposed to read.
+		 */
+		@Nullable
+		Object getResultAndReset();
+	}
+
+	/**
+	 * Adapts a {@link Map} to the interface of a {@literal Collection<Map.Entry<Object, Object>>}.
+	 * 
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	private static class MapAdapter extends AbstractCollection<Map.Entry<Object, Object>> {
+
+		private final Map<Object, Object> map = new HashMap<>();
+
+		@Override
+		public Iterator<Map.Entry<Object, Object>> iterator() {
+			return map.entrySet().iterator();
+		}
+
+		@Override
+		public int size() {
+			return map.size();
+		}
+
+		@Override
+		public boolean add(Map.Entry<Object, Object> entry) {
+
+			map.put(entry.getKey(), entry.getValue());
+			return true;
+		}
+	}
+
+	/**
+	 * Adapts a {@link List} to the interface of a {@literal Collection<Map.Entry<Object, Object>>}.
+	 *
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	private static class ListAdapter extends AbstractCollection<Map.Entry<Object, Object>> {
+
+		private final List<Object> list = new ArrayList<>();
+
+		@Override
+		public Iterator<Map.Entry<Object, Object>> iterator() {
+			throw new UnsupportedOperationException("Do we need this?");
+		}
+
+		@Override
+		public int size() {
+			return list.size();
+		}
+
+		@Override
+		public boolean add(Map.Entry<Object, Object> entry) {
+
+			Integer index = (Integer) entry.getKey();
+			while (index >= list.size()) {
+				list.add(null);
+			}
+			list.set(index, entry.getValue());
+			return true;
+		}
+	}
+
+	/**
+	 * A {@link Reader} for reading entities.
+	 * 
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	private class EntityReader implements Reader {
+
+		/**
+		 * Debugging the recursive structure of {@link Reader} instances can become a little mind bending. Giving each
+		 * {@literal Reader} a descriptive name helps with that.
+		 */
+		private final String name;
+
+		private final AggregatePath basePath;
+		private final CachingResultSet crs;
+
+		private final EntityInstantiator instantiator;
+		@Nullable private final String idColumn;
+
+		private ResultSetParameterValueProvider valueProvider;
+		private boolean result;
+
+		Object oldId = null;
+
+		private EntityReader(AggregatePath basePath, CachingResultSet crs) {
+			this(basePath, crs, null);
+		}
+
+		private EntityReader(AggregatePath basePath, CachingResultSet crs, @Nullable String keyColumn) {
+
+			this.basePath = basePath;
+			this.crs = crs;
+
+			RelationalPersistentEntity<?> entity = basePath.isRoot() ? rootEntity : basePath.getRequiredLeafEntity();
+			instantiator = converter.getEntityInstantiators().getInstantiatorFor(entity);
+
+			idColumn = entity.hasIdProperty() ? propertyToColumn.column(basePath.append(entity.getRequiredIdProperty()))
+					: keyColumn;
+
+			reset();
+
+			name = "EntityReader for " + (basePath.isRoot() ? "<root>" : basePath.toDotPath());
+		}
+
+		@Override
+		public void read() {
+
+			if (idColumn != null && oldId == null) {
+				oldId = crs.getObject(idColumn);
+			}
+
+			valueProvider.readValues();
+			if (idColumn == null) {
+				result = true;
+			} else {
+				Object peekedId = crs.peek(idColumn);
+				if (peekedId == null || !peekedId.equals(oldId)) {
+
+					result = true;
+					oldId = peekedId;
+				}
+			}
+		}
+
+		@Override
+		public boolean hasResult() {
+			return result;
+		}
+
+		@Override
+		@Nullable
+		public Object getResultAndReset() {
+
+			try {
+				return hydrateInstance(instantiator, valueProvider, valueProvider.baseEntity);
+			} finally {
+
+				reset();
+			}
+		}
+
+		private void reset() {
+
+			valueProvider = new ResultSetParameterValueProvider(crs, basePath);
+			result = false;
+		}
+
+		@Override
+		public String toString() {
+			return name;
+		}
+	}
+
+	/**
+	 * A {@link Reader} for reading collections of entities.
+	 * 
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	class CollectionReader implements Reader {
+
+		// debugging only
+		private final String name;
+
+		private final Supplier<Collection> collectionInitializer;
+		private final Reader entityReader;
+
+		private Collection result;
+
+		private static Supplier<Collection> collectionInitializerFor(AggregatePath path) {
+
+			RelationalPersistentProperty property = path.getRequiredLeafProperty();
+			if (List.class.isAssignableFrom(property.getType())) {
+				return ListAdapter::new;
+			} else if (property.isMap()) {
+				return MapAdapter::new;
+			} else {
+				return HashSet::new;
+			}
+		}
+
+		private CollectionReader(AggregatePath basePath, CachingResultSet crs) {
+
+			this.collectionInitializer = collectionInitializerFor(basePath);
+
+			String keyColumn = null;
+			final RelationalPersistentProperty property = basePath.getRequiredLeafProperty();
+			if (property.isMap() || List.class.isAssignableFrom(basePath.getRequiredLeafProperty().getType())) {
+				keyColumn = propertyToColumn.keyColumn(basePath);
+			}
+
+			if (property.isQualified()) {
+				this.entityReader = new EntryReader(basePath, crs, keyColumn, property.getQualifierColumnType());
+			} else {
+				this.entityReader = new EntityReader(basePath, crs, keyColumn);
+			}
+			reset();
+			name = "Reader for " + basePath.toDotPath();
+		}
+
+		private CollectionReader(CachingResultSet crs) {
+
+			this.collectionInitializer = ArrayList::new;
+			this.entityReader = new EntityReader(context.getAggregatePath(rootEntity), crs);
+			reset();
+
+			name = "Collectionreader for <root>";
+
+		}
+
+		@Override
+		public void read() {
+
+			entityReader.read();
+			if (entityReader.hasResult()) {
+				result.add(entityReader.getResultAndReset());
+			}
+		}
+
+		@Override
+		public boolean hasResult() {
+			return false;
+		}
+
+		@Override
+		public Object getResultAndReset() {
+
+			try {
+				if (result instanceof MapAdapter) {
+					return ((MapAdapter) result).map;
+				}
+				if (result instanceof ListAdapter) {
+					return ((ListAdapter) result).list;
+				}
+				return result;
+			} finally {
+				reset();
+			}
+		}
+
+		private void reset() {
+			result = collectionInitializer.get();
+		}
+
+		@Override
+		public String toString() {
+			return name;
+		}
+	}
+
+	/**
+	 * A {@link Reader} for reading collection entries. Most of the work is done by an {@link EntityReader}, but a
+	 * additional key column might get read. The result is
+	 * 
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	private class EntryReader implements Reader {
+
+		final EntityReader delegate;
+		final String keyColumn;
+		private final TypeInformation<?> keyColumnType;
+
+		Object key;
+
+		EntryReader(AggregatePath basePath, CachingResultSet crs, String keyColumn, Class<?> keyColumnType) {
+
+			this.keyColumnType = TypeInformation.of(keyColumnType);
+			this.delegate = new EntityReader(basePath, crs, keyColumn);
+			this.keyColumn = keyColumn;
+		}
+
+		@Override
+		public void read() {
+
+			if (key == null) {
+				Object unconvertedKeyObject = delegate.crs.getObject(keyColumn);
+				key = converter.readValue(unconvertedKeyObject, keyColumnType);
+			}
+			delegate.read();
+		}
+
+		@Override
+		public boolean hasResult() {
+			return delegate.hasResult();
+		}
+
+		@Override
+		public Object getResultAndReset() {
+
+			try {
+				return new AbstractMap.SimpleEntry<>(key, delegate.getResultAndReset());
+			} finally {
+				key = null;
+			}
+		}
+	}
+
+	/**
+	 * A {@link ParameterValueProvider} that provided the values for an entity from a continues set of rows in a {@link ResultSet}. These might be referenced entities or collections of such entities. {@link ResultSet}.
+	 * 
+	 * @since 3.2
+	 * @author Jens Schauder
+	 */
+	private class ResultSetParameterValueProvider implements ParameterValueProvider<RelationalPersistentProperty> {
+
+		private final CachingResultSet rs;
+		/**
+		 * The path which is used to determine columnNames
+		 */
+		private final AggregatePath basePath;
+		private final RelationalPersistentEntity<?> baseEntity;
+
+		/**
+		 * Holds all the values for the entity, either directly or in the form of an appropriate {@link Reader}.
+		 */
+		private final Map<RelationalPersistentProperty, Object> aggregatedValues = new HashMap<>();
+
+		ResultSetParameterValueProvider(CachingResultSet rs, AggregatePath basePath) {
+
+			this.rs = rs;
+			this.basePath = basePath;
+			this.baseEntity = basePath.isRoot() ? rootEntity
+					: context.getRequiredPersistentEntity(basePath.getRequiredLeafProperty().getActualType());
+		}
+
+		@SuppressWarnings("unchecked")
+		@Override
+		@Nullable
+		public <S> S getParameterValue(Parameter<S, RelationalPersistentProperty> parameter) {
+
+			return (S) getValue(baseEntity.getRequiredPersistentProperty(parameter.getName()));
+		}
+
+		@Nullable
+		private Object getValue(RelationalPersistentProperty property) {
+
+			Object value = aggregatedValues.get(property);
+
+			if (value instanceof Reader) {
+				return ((Reader) value).getResultAndReset();
+			}
+
+			value = converter.readValue(value, property.getTypeInformation());
+
+			return value;
+		}
+
+		/**
+		 * read values for all collection like properties and aggregate them in a collection.
+		 */
+		void readValues() {
+			baseEntity.forEach(this::readValue);
+		}
+
+		private void readValue(RelationalPersistentProperty p) {
+
+			if (p.isEntity()) {
+
+				Reader reader = null;
+
+				if (p.isCollectionLike() || p.isMap()) { // even when there are no values we still want a (empty) collection.
+
+					reader = (Reader) aggregatedValues.computeIfAbsent(p, pp -> new CollectionReader(basePath.append(pp), rs));
+				}
+				if (getIndicatorOf(p) != null) {
+
+					if (!(p.isCollectionLike() || p.isMap())) { // for single entities we want a null entity instead of on filled
+																											// with null values.
+
+						reader = (Reader) aggregatedValues.computeIfAbsent(p, pp -> new EntityReader(basePath.append(pp), rs));
+					}
+
+					Assert.state(reader != null, "reader must not be null");
+
+					reader.read();
+				}
+			} else {
+				aggregatedValues.computeIfAbsent(p, this::getObject);
+			}
+		}
+
+		@Nullable
+		private Object getIndicatorOf(RelationalPersistentProperty p) {
+			if (p.isMap() || List.class.isAssignableFrom(p.getType())) {
+				return rs.getObject(getKeyName(p));
+			}
+
+			if (p.isEmbedded()) {
+				return true;
+			}
+
+			return rs.getObject(getColumnName(p));
+		}
+
+		/**
+		 * Obtain a single columnValue from the resultset without throwing an exception. If the column does not exist a null
+		 * value is returned. Does not instantiate complex objects.
+		 *
+		 * @param property
+		 * @return
+		 */
+		@Nullable
+		private Object getObject(RelationalPersistentProperty property) {
+			return rs.getObject(getColumnName(property));
+		}
+
+		/**
+		 * converts a property into a column name representing that property.
+		 *
+		 * @param property
+		 * @return
+		 */
+		private String getColumnName(RelationalPersistentProperty property) {
+
+			return propertyToColumn.column(basePath.append(property));
+		}
+
+		private String getKeyName(RelationalPersistentProperty property) {
+
+			return propertyToColumn.keyColumn(basePath.append(property));
+		}
+
+		private boolean hasValue() {
+
+			for (Object value : aggregatedValues.values()) {
+				if (value != null) {
+					return true;
+				}
+			}
+			return false;
+		}
+	}
+}

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

@@ -394,6 +394,7 @@ public class BasicJdbcConverter extends BasicRelationalConverter implements Jdbc
 		}
 
 		private <S> ReadingContext<S> extendBy(RelationalPersistentProperty property) {
+
 			return new ReadingContext<>(
 					(RelationalPersistentEntity<S>) getMappingContext().getRequiredPersistentEntity(property.getActualType()),
 					rootPath.append(property), path.append(property), identifier, key,

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

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.convert;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Despite its name not really a {@link ResultSet}, but it offers the part of the {@literal ResultSet} API that is used
+ * by {@link AggregateReader}. It allows peeking in the next row of a ResultSet by caching one row of the ResultSet.
+ *
+ * @since 3.2
+ * @author Jens Schauder
+ */
+class CachingResultSet {
+
+	private final ResultSetAccessor accessor;
+	private final ResultSet resultSet;
+	private Cache cache;
+
+	CachingResultSet(ResultSet resultSet) {
+
+		this.accessor = new ResultSetAccessor(resultSet);
+		this.resultSet = resultSet;
+	}
+
+	public boolean next() {
+
+		if (isPeeking()) {
+
+			final boolean next = cache.next;
+			cache = null;
+			return next;
+		}
+
+		try {
+			return resultSet.next();
+		} catch (SQLException e) {
+			throw new RuntimeException("Failed to advance CachingResultSet", e);
+		}
+	}
+
+	@Nullable
+	public Object getObject(String columnLabel) {
+
+		Object returnValue;
+		if (isPeeking()) {
+			returnValue = cache.values.get(columnLabel);
+		} else {
+			returnValue = safeGetFromDelegate(columnLabel);
+		}
+
+		return returnValue;
+	}
+
+	@Nullable
+	Object peek(String columnLabel) {
+
+		if (!isPeeking()) {
+			createCache();
+		}
+
+		if (!cache.next) {
+			return null;
+		}
+
+		return safeGetFromDelegate(columnLabel);
+	}
+
+	@Nullable
+	private Object safeGetFromDelegate(String columnLabel) {
+		return accessor.getObject(columnLabel);
+	}
+
+	private void createCache() {
+		cache = new Cache();
+
+		try {
+			int columnCount = resultSet.getMetaData().getColumnCount();
+			for (int i = 1; i <= columnCount; i++) {
+				// at least some databases return lower case labels although rs.getObject(UPPERCASE_LABEL) returns the expected
+				// value. The aliases we use happen to be uppercase. So we transform everything to upper case.
+				cache.add(resultSet.getMetaData().getColumnLabel(i).toLowerCase(),
+						accessor.getObject(resultSet.getMetaData().getColumnLabel(i)));
+			}
+
+			cache.next = resultSet.next();
+		} catch (SQLException se) {
+			throw new RuntimeException("Can't cache result set data", se);
+		}
+
+	}
+
+	private boolean isPeeking() {
+		return cache != null;
+	}
+
+	private static class Cache {
+
+		boolean next;
+		Map<String, Object> values = new HashMap<>();
+
+		void add(String columnName, Object value) {
+			values.put(columnName, value);
+		}
+	}
+}

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

@@ -29,7 +29,6 @@ import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.conversion.IdValueSource;
 import org.springframework.data.relational.core.mapping.AggregatePath;
-import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
@@ -69,6 +68,7 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 	private final NamedParameterJdbcOperations operations;
 	private final SqlParametersFactory sqlParametersFactory;
 	private final InsertStrategyFactory insertStrategyFactory;
+	private final ReadingDataAccessStrategy singleSelectDelegate;
 
 	/**
 	 * Creates a {@link DefaultDataAccessStrategy}
@@ -96,6 +96,7 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 		this.operations = operations;
 		this.sqlParametersFactory = sqlParametersFactory;
 		this.insertStrategyFactory = insertStrategyFactory;
+		this.singleSelectDelegate = new SingleQueryDataAccessStrategy(context, sqlGeneratorSource.getDialect(), converter, operations);
 	}
 
 	@Override
@@ -260,6 +261,10 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 	@Override
 	public <T> T findById(Object id, Class<T> domainType) {
 
+		if (isSingleSelectQuerySupported(domainType)) {
+			return singleSelectDelegate.findById(id, domainType);
+		}
+
 		String findOneSql = sql(domainType).getFindOne();
 		SqlIdentifierParameterSource parameter = sqlParametersFactory.forQueryById(id, domainType, ID_SQL_PARAMETER);
 
@@ -272,6 +277,11 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 
 	@Override
 	public <T> Iterable<T> findAll(Class<T> domainType) {
+
+		if (isSingleSelectQuerySupported(domainType)){
+			return singleSelectDelegate.findAll(domainType);
+		}
+
 		return operations.query(sql(domainType).getFindAll(), getEntityRowMapper(domainType));
 	}
 
@@ -282,10 +292,12 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 			return Collections.emptyList();
 		}
 
+		if (isSingleSelectQuerySupported(domainType)){
+			return singleSelectDelegate.findAllById(ids, domainType);
+		}
+
 		SqlParameterSource parameterSource = sqlParametersFactory.forQueryByIds(ids, domainType);
-
 		String findAllInListSql = sql(domainType).getFindAllInList();
-
 		return operations.query(findAllInListSql, parameterSource, getEntityRowMapper(domainType));
 	}
 
@@ -430,4 +442,40 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 
 		return baseProperty.getOwner().getType();
 	}
+
+	private boolean isSingleSelectQuerySupported(Class<?> entityType) {
+
+		return context.isSingleQueryLoadingEnabled() && sqlGeneratorSource.getDialect().supportsSingleQueryLoading()//
+				&& entityQualifiesForSingleSelectQuery(entityType);
+	}
+
+	private boolean entityQualifiesForSingleSelectQuery(Class<?> entityType) {
+
+		boolean referenceFound = false;
+		for (PersistentPropertyPath<RelationalPersistentProperty> path : context.findPersistentPropertyPaths(entityType, __ -> true)) {
+			RelationalPersistentProperty property = path.getLeafProperty();
+			if (property.isEntity()) {
+
+				// embedded entities are currently not supported
+				if (property.isEmbedded()) {
+					return false;
+				}
+
+				// only a single reference is currently supported
+				if (referenceFound) {
+					return false;
+				}
+
+				referenceFound = true;
+			}
+
+			// AggregateReferences aren't supported yet
+			if (property.isAssociation()) {
+				return false;
+			}
+		}
+		return true;
+
+	}
+
 }

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.jdbc.core.convert;
+
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.AggregatePath;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+
+/**
+ * A mapping between {@link PersistentPropertyPath} and column names of a query. Column names are intentionally
+ * represented by {@link String} values, since this is what a {@link java.sql.ResultSet} uses, and since all the query
+ * columns should be aliases there is no need for quoting or similar as provided by
+ * {@link org.springframework.data.relational.core.sql.SqlIdentifier}.
+ * 
+ * @since 3.2
+ * @author Jens Schauder
+ */
+public interface PathToColumnMapping {
+
+	String column(AggregatePath path);
+
+	String keyColumn(AggregatePath path);
+}

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

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.jdbc.core.convert;
+
+import java.util.Optional;
+
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.relational.core.query.Query;
+import org.springframework.lang.Nullable;
+
+/**
+ * The finding methods of a {@link DataAccessStrategy}.
+ *
+ * @since 3.2
+ * @author Jens Schauder
+ */
+interface ReadingDataAccessStrategy {
+	/**
+	 * Loads a single entity identified by type and id.
+	 *
+	 * @param id the id of the entity to load. Must not be {@code null}.
+	 * @param domainType the domain type of the entity. Must not be {@code null}.
+	 * @param <T> the type of the entity.
+	 * @return Might return {@code null}.
+	 */
+	@Nullable
+	<T> T findById(Object id, Class<T> domainType);
+
+	/**
+	 * Loads all entities of the given type.
+	 *
+	 * @param domainType the type of entities to load. Must not be {@code null}.
+	 * @param <T> the type of entities to load.
+	 * @return Guaranteed to be not {@code null}.
+	 */
+	<T> Iterable<T> findAll(Class<T> domainType);
+
+	/**
+	 * Loads all entities that match one of the ids passed as an argument. It is not guaranteed that the number of ids
+	 * passed in matches the number of entities returned.
+	 *
+	 * @param ids the Ids of the entities to load. Must not be {@code null}.
+	 * @param domainType the type of entities to load. Must not be {@code null}.
+	 * @param <T> type of entities to load.
+	 * @return the loaded entities. Guaranteed to be not {@code null}.
+	 */
+	<T> Iterable<T> findAllById(Iterable<?> ids, Class<T> domainType);
+
+	/**
+	 * Loads all entities of the given type, sorted.
+	 *
+	 * @param domainType the type of entities to load. Must not be {@code null}.
+	 * @param <T> the type of entities to load.
+	 * @param sort the sorting information. Must not be {@code null}.
+	 * @return Guaranteed to be not {@code null}.
+	 * @since 2.0
+	 */
+	<T> Iterable<T> findAll(Class<T> domainType, Sort sort);
+
+	/**
+	 * Loads all entities of the given type, paged and sorted.
+	 *
+	 * @param domainType the type of entities to load. Must not be {@code null}.
+	 * @param <T> the type of entities to load.
+	 * @param pageable the pagination information. Must not be {@code null}.
+	 * @return Guaranteed to be not {@code null}.
+	 * @since 2.0
+	 */
+	<T> Iterable<T> findAll(Class<T> domainType, Pageable pageable);
+
+	/**
+	 * Execute a {@code SELECT} query and convert the resulting item to an entity ensuring exactly one result.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param domainType the type of entities. Must not be {@code null}.
+	 * @return exactly one result or {@link Optional#empty()} if no match found.
+	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+	 * @since 3.0
+	 */
+	<T> Optional<T> findOne(Query query, Class<T> domainType);
+
+	/**
+	 * Execute a {@code SELECT} query and convert the resulting items to a {@link Iterable}.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param domainType the type of entities. Must not be {@code null}.
+	 * @return a non-null list with all the matching results.
+	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+	 * @since 3.0
+	 */
+	<T> Iterable<T> findAll(Query query, Class<T> domainType);
+
+	/**
+	 * Execute a {@code SELECT} query and convert the resulting items to a {@link Iterable}. Applies the {@link Pageable}
+	 * to the result.
+	 *
+	 * @param query must not be {@literal null}.
+	 * @param domainType the type of entities. Must not be {@literal  null}.
+	 * @param pageable the pagination that should be applied. Must not be {@literal null}.
+	 * @return a non-null list with all the matching results.
+	 * @throws org.springframework.dao.IncorrectResultSizeDataAccessException if more than one match found.
+	 * @since 3.0
+	 */
+	<T> Iterable<T> findAll(Query query, Class<T> domainType, Pageable pageable);
+}

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.jdbc.core.convert;
+
+import java.util.Optional;
+
+import org.springframework.data.domain.Pageable;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.relational.core.dialect.Dialect;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.query.Query;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+
+/**
+ * A {@link ReadingDataAccessStrategy} that uses an {@link AggregateReader} to load entities with a single query.
+ * 
+ * @since 3.2
+ * @author Jens Schauder
+ */
+public class SingleQueryDataAccessStrategy implements ReadingDataAccessStrategy {
+	private final AggregateReaderFactory readerFactory;
+	private final RelationalMappingContext mappingContext;
+
+	public SingleQueryDataAccessStrategy(RelationalMappingContext mappingContext, Dialect dialect,
+			JdbcConverter converter, NamedParameterJdbcOperations jdbcTemplate) {
+
+		this.mappingContext = mappingContext;
+		this.readerFactory = new AggregateReaderFactory(mappingContext, dialect, converter, jdbcTemplate);
+		;
+	}
+
+	@Override
+	public <T> T findById(Object id, Class<T> domainType) {
+		return getReader(domainType).findById(id);
+	}
+
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType) {
+		return getReader(domainType).findAll();
+	}
+
+	@Override
+	public <T> Iterable<T> findAllById(Iterable<?> ids, Class<T> domainType) {
+		return getReader(domainType).findAllById(ids);
+	}
+
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Sort sort) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <T> Iterable<T> findAll(Class<T> domainType, Pageable pageable) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <T> Optional<T> findOne(Query query, Class<T> domainType) {
+		return Optional.empty();
+	}
+
+	@Override
+	public <T> Iterable<T> findAll(Query query, Class<T> domainType) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public <T> Iterable<T> findAll(Query query, Class<T> domainType, Pageable pageable) {
+		throw new UnsupportedOperationException();
+	}
+
+	private <T> AggregateReader<T> getReader(Class<T> domainType) {
+
+		RelationalPersistentEntity<T> persistentEntity = (RelationalPersistentEntity<T>) mappingContext
+				.getRequiredPersistentEntity(domainType);
+		return readerFactory.createAggregateReaderFor(persistentEntity);
+	}
+}