JdbcAggregateTemplate honors columns specified in query.

If no columns are given, all columns are selected by default.

If columns are specified, only these are selected.
Joins normally triggered by columns from 1:1 relationships are not implemented, and the corresponding columns don't get loaded and can't be specified in a query.

Limiting columns is not supported for single query loading.

Closes #1803

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

@@ -507,33 +507,50 @@ private String createFindAllSql() {
 	}
 
 	private SelectBuilder.SelectWhere selectBuilder() {
-		return selectBuilder(Collections.emptyList());
+		return selectBuilder(Collections.emptyList(), null);
+	}
+
+
+	private SelectBuilder.SelectWhere selectBuilder(Query query) {
+		return selectBuilder(Collections.emptyList(), query);
 	}
 
 	private SelectBuilder.SelectWhere selectBuilder(Collection<SqlIdentifier> keyColumns) {
+		return selectBuilder(keyColumns, null);
+	}
+
+	private SelectBuilder.SelectWhere selectBuilder(Collection<SqlIdentifier> keyColumns, @Nullable Query query) {
 
 		Table table = getTable();
 
 		Set<Expression> columnExpressions = new LinkedHashSet<>();
 
 		List<Join> joinTables = new ArrayList<>();
-		for (PersistentPropertyPath<RelationalPersistentProperty> path : mappingContext
-				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
 
-			AggregatePath extPath = mappingContext.getAggregatePath(path);
-
-			// add a join if necessary
-			Join join = getJoin(extPath);
-			if (join != null) {
-				joinTables.add(join);
+		if (query != null && !query.getColumns().isEmpty()) {
+			for (SqlIdentifier columnName : query.getColumns()) {
+				columnExpressions.add(Column.create(columnName, table));
 			}
+		} else {
+			for (PersistentPropertyPath<RelationalPersistentProperty> path : mappingContext
+					.findPersistentPropertyPaths(entity.getType(), p -> true)) {
+
+				AggregatePath extPath = mappingContext.getAggregatePath(path);
+
+				// add a join if necessary
+				Join join = getJoin(extPath);
+				if (join != null) {
+					joinTables.add(join);
+				}
 
-			Column column = getColumn(extPath);
-			if (column != null) {
-				columnExpressions.add(column);
+				Column column = getColumn(extPath);
+				if (column != null) {
+					columnExpressions.add(column);
+				}
 			}
 		}
 
+
 		for (SqlIdentifier keyColumn : keyColumns) {
 			columnExpressions.add(table.column(keyColumn).as(keyColumn));
 		}
@@ -876,14 +893,15 @@ public String selectByQuery(Query query, MapSqlParameterSource parameterSource)
 
 		Assert.notNull(parameterSource, "parameterSource must not be null");
 
-		SelectBuilder.SelectWhere selectBuilder = selectBuilder();
+		SelectBuilder.SelectWhere selectBuilder = selectBuilder(query);
 
 		Select select = applyQueryOnSelect(query, parameterSource, selectBuilder) //
 				.build();
 
 		return render(select);
 	}
 
+
 	/**
 	 * Constructs a single sql query that performs select based on the provided query and pagination information.
 	 * Additional the bindings for the where clause are stored after execution into the <code>parameterSource</code>

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

@@ -29,6 +29,7 @@
 import java.util.stream.IntStream;
 
 import org.assertj.core.api.SoftAssertions;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationEventPublisher;
@@ -235,7 +236,25 @@ void findAllByQuery() {
 		Query query = Query.query(criteria);
 		Iterable<SimpleListParent> reloadedById = template.findAll(query, SimpleListParent.class);
 
-		assertThat(reloadedById).extracting(e -> e.id, e -> e.content.size()).containsExactly(tuple(two.id, 2));
+		assertThat(reloadedById) //
+				.extracting(e -> e.id, e-> e.name, e -> e.content.size()) //
+				.containsExactly(tuple(two.id, two.name, 2));
+	}
+
+	@Test // GH-1803
+	void findAllByQueryWithColumns() {
+
+		template.save(SimpleListParent.of("one", "one_1"));
+		SimpleListParent two = template.save(SimpleListParent.of("two", "two_1", "two_2"));
+		template.save(SimpleListParent.of("three", "three_1", "three_2", "three_3"));
+
+		CriteriaDefinition criteria = CriteriaDefinition.from(Criteria.where("id").is(two.id));
+		Query query = Query.query(criteria).columns("id");
+		Iterable<SimpleListParent> reloadedById = template.findAll(query, SimpleListParent.class);
+
+		assertThat(reloadedById) //
+				.extracting(e -> e.id, e-> e.name, e -> e.content.size()) //
+				.containsExactly(tuple(two.id, null, 2));
 	}
 
 	@Test // GH-1601
@@ -2240,5 +2259,10 @@ static class JdbcAggregateTemplateIntegrationTests extends AbstractJdbcAggregate
 	static class JdbcAggregateTemplateSingleQueryLoadingIntegrationTests
 			extends AbstractJdbcAggregateTemplateIntegrationTests {
 
+		@Disabled
+		@Override
+		void findAllByQueryWithColumns() {
+			super.findAllByQueryWithColumns();
+		}
 	}
 }

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

@@ -358,6 +358,19 @@ void selectByQuery() {
 		);
 	}
 
+	@Test // GH-1803
+	void selectByQueryWithColumnLimit() {
+
+		Query query = Query.empty().columns("alpha", "beta", "gamma");
+
+		String sql = sqlGenerator.selectByQuery(query, new MapSqlParameterSource());
+
+		assertThat(sql).contains( //
+				"SELECT dummy_entity.alpha, dummy_entity.beta, dummy_entity.gamma", //
+				"FROM dummy_entity" //
+		);
+	}
+
 	@Test // GH-1919
 	void selectBySortedQuery() {
 
@@ -375,7 +388,8 @@ void selectBySortedQuery() {
 				"ORDER BY dummy_entity.id1 ASC" //
 		);
 		assertThat(sql).containsOnlyOnce("LEFT OUTER JOIN referenced_entity ref ON ref.dummy_entity = dummy_entity.id1");
-		assertThat(sql).containsOnlyOnce("LEFT OUTER JOIN second_level_referenced_entity ref_further ON ref_further.referenced_entity = ref.x_l1id");
+		assertThat(sql).containsOnlyOnce(
+				"LEFT OUTER JOIN second_level_referenced_entity ref_further ON ref_further.referenced_entity = ref.x_l1id");
 	}
 
 	@Test // DATAJDBC-131, DATAJDBC-111