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 off Single Query Loading call `RelationalMappingContext.setSingleQueryLoadingEnabled(false)`.

Closes #1446
See #1450
See #1445

ci/accept-third-party-license.sh

@@ -1,7 +1,7 @@
 #!/bin/sh
 
 {
-  echo "mcr.microsoft.com/mssql/server:2019-CU16-ubuntu-20.04"
+  echo "mcr.microsoft.com/mssql/server:2022-CU5-ubuntu-20.04"
   echo "ibmcom/db2:11.5.7.0a"
   echo "harbor-repo.vmware.com/dockerhub-proxy-cache/ibmcom/db2:11.5.7.0a"
 } > spring-data-jdbc/src/test/resources/container-license-acceptance.txt

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);
+	}
+}