DATAJDBC-340 - Using new SQL generation DSL.

SqlGenerator basically got rewritten using the new SQL rendering API.

The SQL rendering API got some minor improvements in the process.

The whole generation process is now based on paths instead of properties which makes the actual logic of the construction much more obvious.

Original pull request: #147.

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

@@ -0,0 +1,300 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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;
+
+import java.util.Objects;
+
+import org.springframework.data.mapping.PersistentPropertyPath;
+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.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * A wrapper around a {@link org.springframework.data.mapping.PersistentPropertyPath} for making common operations
+ * available used in SQL generation.
+ *
+ * @author Jens Schauder
+ * @since 1.1
+ */
+class PersistentPropertyPathExtension {
+
+	private final RelationalPersistentEntity<?> entity;
+	private final PersistentPropertyPath<RelationalPersistentProperty> path;
+	private final RelationalMappingContext context;
+
+	PersistentPropertyPathExtension(RelationalMappingContext context, RelationalPersistentEntity<?> entity) {
+
+		Assert.notNull(context, "Context must not be null.");
+		Assert.notNull(entity, "Entity must not be null.");
+
+		this.context = context;
+		this.entity = entity;
+		this.path = null;
+	}
+
+	PersistentPropertyPathExtension(RelationalMappingContext context,
+			PersistentPropertyPath<RelationalPersistentProperty> path) {
+
+		Assert.notNull(context, "Context must not be null.");
+		Assert.notNull(path, "Path must not be null.");
+		Assert.isTrue(!path.isEmpty(), "Path must not be empty.");
+
+		this.context = context;
+		this.entity = Objects.requireNonNull(path.getBaseProperty()).getOwner();
+		this.path = path;
+	}
+
+	/**
+	 * Returns {@literal true} exactly when the path is non empty and the leaf property an embedded one.
+	 *
+	 * @return if the leaf property is embedded.
+	 */
+	boolean isEmbedded() {
+		return path != null && path.getRequiredLeafProperty().isEmbedded();
+	}
+
+	/**
+	 * Returns the path that has the same beginning but is one segment shorter than this path.
+	 *
+	 * @return the parent path. Guaranteed to be not {@literal null}.
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	PersistentPropertyPathExtension getParentPath() {
+
+		if (path == null) {
+			throw new IllegalStateException("The parent path of a root path is not defined.");
+		}
+
+		if (path.getLength() == 1) {
+			return new PersistentPropertyPathExtension(context, entity);
+		}
+
+		return new PersistentPropertyPathExtension(context, path.getParentPath());
+	}
+
+	/**
+	 * Returns {@literal true} if there are multiple values for this path, i.e. if the path contains at least one element
+	 * that is a collection and array or a map.
+	 *
+	 * @return {@literal true} if the path contains a multivalued element.
+	 */
+	boolean isMultiValued() {
+
+		return path != null && //
+				(path.getRequiredLeafProperty().isCollectionLike() //
+						|| path.getRequiredLeafProperty().isQualified() //
+						|| getParentPath().isMultiValued() //
+				);
+	}
+
+	/**
+	 * The {@link RelationalPersistentEntity} associated with the leaf of this path.
+	 *
+	 * @return Might return {@literal null} when called on a path that does not represent an entity.
+	 */
+	@Nullable
+	RelationalPersistentEntity<?> getLeafEntity() {
+		return path == null ? entity : context.getPersistentEntity(path.getRequiredLeafProperty().getActualType());
+	}
+
+	/**
+	 * @return {@literal true} when this is an empty path or the path references an entity.
+	 */
+	boolean isEntity() {
+		return path == null || path.getRequiredLeafProperty().isEntity();
+	}
+
+	/**
+	 * @return {@literal true} when this is references a {@link java.util.List} or {@link java.util.Map}.
+	 */
+	boolean isQualified() {
+		return path != null && path.getRequiredLeafProperty().isQualified();
+	}
+
+	/**
+	 * @return {@literal true} when this is references a {@link java.util.Collection} or an array.
+	 */
+	boolean isCollectionLike() {
+		return path != null && path.getRequiredLeafProperty().isCollectionLike();
+	}
+
+	/**
+	 * The name of the column used to reference the id in the parent table.
+	 *
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	String getReverseColumnName() {
+
+		return path.getRequiredLeafProperty().getReverseColumnName();
+	}
+
+	/**
+	 * The alias used in select for the column used to reference the id in the parent table.
+	 *
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	String getReverseColumnNameAlias() {
+
+		return prefixWithTableAlias(getReverseColumnName());
+	}
+
+	/**
+	 * The name of the column used to represent this property in the database.
+	 *
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	String getColumnName() {
+
+		return assembleColumnName(path.getRequiredLeafProperty().getColumnName());
+	}
+
+	/**
+	 * The alias for the column used to represent this property in the database.
+	 *
+	 * @throws IllegalStateException when called on an empty path.
+	 */
+	String getColumnAlias() {
+
+		return prefixWithTableAlias(getColumnName());
+	}
+
+	/**
+	 * @return {@literal true} if this path represents an entity which has an Id attribute.
+	 */
+	boolean hasIdProperty() {
+
+		RelationalPersistentEntity<?> leafEntity = getLeafEntity();
+		return leafEntity != null && leafEntity.hasIdProperty();
+	}
+
+	PersistentPropertyPathExtension getIdDefiningParentPath() {
+
+		PersistentPropertyPathExtension parent = getParentPath();
+		if (parent.path == null) {
+			return parent;
+		}
+		if (parent.isEmbedded()) {
+			return getParentPath().getIdDefiningParentPath();
+		}
+		return parent;
+	}
+
+	/**
+	 * The name of the table this path is tied to or of the longest ancestor path that is actually tied to a table.
+	 *
+	 * @return the name of the table. Guaranteed to be not {@literal null}.
+	 */
+	String getTableName() {
+		return getTableOwningAncestor().getRequiredLeafEntity().getTableName();
+	}
+
+	/**
+	 * The alias used for the table on which this path is based.
+	 *
+	 * @return a table alias, {@literal null} if the table owning path is the empty path.
+	 */
+	@Nullable
+	String getTableAlias() {
+
+		PersistentPropertyPathExtension tableOwner = getTableOwningAncestor();
+		if (tableOwner.path == null) {
+			return null;
+		}
+
+		return tableOwner.assembleTableAlias();
+	}
+
+	/**
+	 * The column name of the id column of the ancestor path that represents an actual table.
+	 */
+	String getIdColumnName() {
+		return getTableOwningAncestor().getRequiredLeafEntity().getIdColumn();
+	}
+
+	/**
+	 * If the table owning ancestor has an id the column name of that id property is returned. Otherwise the reverse
+	 * column is returned.
+	 */
+	String getEffectiveIdColumnName() {
+
+		PersistentPropertyPathExtension owner = getTableOwningAncestor();
+		return owner.path == null ? owner.getRequiredLeafEntity().getIdColumn() : owner.getReverseColumnName();
+	}
+
+	/**
+	 * The length of the path.
+	 */
+	int getLength() {
+		return path == null ? 0 : path.getLength();
+	}
+
+	/**
+	 * Finds and returns the longest path with ich identical or an ancestor to the current path and maps directly to a
+	 * table.
+	 *
+	 * @return a path. Guaranteed to be not {@literal null}.
+	 */
+	private PersistentPropertyPathExtension getTableOwningAncestor() {
+
+		if (isEntity() && !isEmbedded()) {
+			return this;
+		}
+		return getParentPath().getTableOwningAncestor();
+	}
+
+	private String assembleTableAlias() {
+
+		RelationalPersistentProperty leafProperty = path.getRequiredLeafProperty();
+		String prefix = isEmbedded() ? leafProperty.getEmbeddedPrefix() : leafProperty.getName();
+
+		if (path.getLength() == 1) {
+			Assert.notNull(prefix, "Prefix mus not be null.");
+			return prefix;
+		}
+
+		PersistentPropertyPathExtension parentPath = getParentPath();
+		return parentPath.isEmbedded() ? parentPath.assembleTableAlias() + prefix
+				: parentPath.assembleTableAlias() + "_" + prefix;
+	}
+
+	private String assembleColumnName(String suffix) {
+
+		if (path.getLength() <= 1) {
+			return suffix;
+		}
+		PersistentPropertyPath<RelationalPersistentProperty> parentPath = path.getParentPath();
+		RelationalPersistentProperty parentLeaf = parentPath.getRequiredLeafProperty();
+		if (!parentLeaf.isEmbedded()) {
+			return suffix;
+		}
+		String embeddedPrefix = parentLeaf.getEmbeddedPrefix();
+		return getParentPath().assembleColumnName(embeddedPrefix + suffix);
+	}
+
+	private RelationalPersistentEntity<?> getRequiredLeafEntity() {
+		return path == null ? entity : context.getRequiredPersistentEntity(path.getRequiredLeafProperty().getActualType());
+	}
+
+
+	private String prefixWithTableAlias(String columnName) {
+
+		String tableAlias = getTableAlias();
+		return tableAlias == null ? columnName : tableAlias + "_" + columnName;
+	}
+
+}

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

@@ -1,335 +0,0 @@
-/*
- * Copyright 2017-2019 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;
-
-import lombok.Builder;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-/**
- * Builder for creating Select-statements. Not intended for general purpose, but only for the needs of the
- * {@link JdbcAggregateTemplate}.
- *
- * @author Jens Schauder
- */
-class SelectBuilder {
-
-	private final List<Column> columns = new ArrayList<>();
-	private final String tableName;
-	private final List<Join> joins = new ArrayList<>();
-	private final List<WhereCondition> conditions = new ArrayList<>();
-
-	/**
-	 * Creates a {@link SelectBuilder} using the given table name.
-	 * 
-	 * @param tableName the table name. Must not be {@code null}.
-	 */
-	SelectBuilder(String tableName) {
-
-		this.tableName = tableName;
-	}
-
-	/**
-	 * Adds a column to the select list.
-	 *
-	 * @param columnSpec a function that specifies the column to add. The passed in {@link Column.ColumnBuilder} allows to
-	 *          specify details like alias and the source table. Must not be {@code null}.
-	 * @return {@code this}.
-	 */
-	SelectBuilder column(Function<Column.ColumnBuilder, Column.ColumnBuilder> columnSpec) {
-
-		columns.add(columnSpec.apply(Column.builder()).build());
-		return this;
-	}
-
-	/**
-	 * Adds a where clause to the select
-	 *
-	 * @param whereSpec a function specifying the details of the where clause by manipulating the passed in
-	 *          {@link WhereConditionBuilder}. Must not be {@code null}.
-	 * @return {@code this}.
-	 */
-	SelectBuilder where(Function<WhereConditionBuilder, WhereConditionBuilder> whereSpec) {
-
-		conditions.add(whereSpec.apply(new WhereConditionBuilder()).build());
-		return this;
-	}
-
-	/**
-	 * Adds a join to the select.
-	 *
-	 * @param joinSpec a function specifying the details of the join by manipulating the passed in
-	 *          {@link Join.JoinBuilder}. Must not be {@code null}.
-	 * @return {@code this}.
-	 */
-	SelectBuilder join(Function<Join.JoinBuilder, Join.JoinBuilder> joinSpec) {
-
-		joins.add(joinSpec.apply(Join.builder()).build());
-		return this;
-	}
-
-	/**
-	 * Builds the actual SQL statement.
-	 *
-	 * @return a SQL statement. Guaranteed to be not {@code null}.
-	 */
-	String build() {
-
-		return selectFrom() + joinClause() + whereClause();
-	}
-
-	private String whereClause() {
-
-		if (conditions.isEmpty()) {
-			return "";
-		}
-
-		return conditions.stream() //
-				.map(WhereCondition::toSql) //
-				.collect(Collectors.joining("AND", " WHERE ", "") //
-				);
-	}
-
-	private String joinClause() {
-		return joins.stream().map(j -> joinTable(j) + joinConditions(j)).collect(Collectors.joining(" "));
-	}
-
-	private String joinTable(Join j) {
-		return String.format("%s JOIN %s AS %s", j.outerJoinModifier(), j.table, j.as);
-	}
-
-	private String joinConditions(Join j) {
-
-		return j.conditions.stream() //
-				.map(w -> String.format("%s %s %s", w.fromExpression, w.operation, w.toExpression)) //
-				.collect(Collectors.joining(" AND ", " ON ", ""));
-	}
-
-	private String selectFrom() {
-
-		return columns.stream() //
-				.map(Column::columnDefinition) //
-				.collect(Collectors.joining(", ", "SELECT ", " FROM " + tableName));
-	}
-
-	static class WhereConditionBuilder {
-
-		private String fromTable;
-		private String fromColumn;
-
-		private String operation = "=";
-		private String expression;
-
-		WhereConditionBuilder() {}
-
-		WhereConditionBuilder eq() {
-
-			this.operation = "=";
-			return this;
-		}
-
-		public WhereConditionBuilder in() {
-
-			this.operation = "in";
-			return this;
-		}
-
-		WhereConditionBuilder tableAlias(String fromTable) {
-
-			this.fromTable = fromTable;
-			return this;
-		}
-
-		WhereConditionBuilder column(String fromColumn) {
-
-			this.fromColumn = fromColumn;
-			return this;
-		}
-
-		WhereConditionBuilder variable(String var) {
-
-			this.expression = ":" + var;
-			return this;
-		}
-
-		WhereCondition build() {
-			return new WhereCondition(fromTable + "." + fromColumn, operation, expression);
-		}
-
-	}
-
-	static class Join {
-
-		private final String table;
-		private final String as;
-		private final Outer outer;
-		private final List<WhereCondition> conditions = new ArrayList<>();
-
-		Join(String table, String as, List<WhereCondition> conditions, Outer outer) {
-
-			this.table = table;
-			this.as = as;
-			this.outer = outer;
-			this.conditions.addAll(conditions);
-		}
-
-		static JoinBuilder builder() {
-			return new JoinBuilder();
-		}
-
-		private String outerJoinModifier() {
-
-			switch (outer) {
-				case NONE:
-					return "";
-				default:
-					return String.format(" %s OUTER", outer.name());
-
-			}
-		}
-
-		public static class JoinBuilder {
-
-			private String table;
-			private String as;
-			private List<WhereCondition> conditions = new ArrayList<>();
-			private Outer outer = Outer.NONE;
-
-			JoinBuilder() {}
-
-			public Join.JoinBuilder table(String table) {
-
-				this.table = table;
-				return this;
-			}
-
-			public Join.JoinBuilder as(String as) {
-
-				this.as = as;
-				return this;
-			}
-
-			WhereConditionBuilder where(String column) {
-				return new WhereConditionBuilder(this, column);
-			}
-
-			private JoinBuilder where(WhereCondition condition) {
-
-				conditions.add(condition);
-				return this;
-			}
-
-			Join build() {
-				return new Join(table, as, conditions, outer);
-			}
-
-			public String toString() {
-				return "org.springframework.data.jdbc.core.SelectBuilder.Join.JoinBuilder(table=" + this.table + ", as="
-						+ this.as + ")";
-			}
-
-			JoinBuilder rightOuter() {
-
-				outer = Outer.RIGHT;
-				return this;
-			}
-
-			JoinBuilder leftOuter() {
-				outer = Outer.LEFT;
-				return this;
-			}
-
-			static class WhereConditionBuilder {
-
-				private final JoinBuilder joinBuilder;
-				private final String fromColumn;
-
-				private String operation = "=";
-
-				WhereConditionBuilder(JoinBuilder joinBuilder, String column) {
-
-					this.joinBuilder = joinBuilder;
-					this.fromColumn = column;
-				}
-
-				WhereConditionBuilder eq() {
-					operation = "=";
-					return this;
-				}
-
-				JoinBuilder column(String table, String column) {
-
-					return joinBuilder.where(new WhereCondition( //
-							joinBuilder.as + "." + fromColumn, //
-							operation, //
-							table + "." + column //
-					));
-
-				}
-
-			}
-
-		}
-
-		private enum Outer {
-			NONE, RIGHT, LEFT
-		}
-	}
-
-	static class WhereCondition {
-
-		private final String operation;
-		private final String fromExpression;
-		private final String toExpression;
-
-		WhereCondition(String fromExpression, String operation, String toExpression) {
-
-			this.fromExpression = fromExpression;
-			this.toExpression = toExpression;
-			this.operation = operation;
-		}
-
-		String toSql() {
-
-			if (operation.equals("in")) {
-				return String.format("%s %s(%s)", fromExpression, operation, toExpression);
-			}
-
-			return String.format("%s %s %s", fromExpression, operation, toExpression);
-		}
-	}
-
-	@Builder
-	static class Column {
-
-		private final String tableAlias;
-		private final String column;
-		private final String as;
-
-		String columnDefinition() {
-			StringBuilder b = new StringBuilder();
-			if (tableAlias != null)
-				b.append(tableAlias).append('.');
-			b.append(column);
-			if (as != null)
-				b.append(" AS ").append(as);
-			return b.toString();
-		}
-	}
-}

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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;
+
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.sql.Column;
+import org.springframework.data.relational.core.sql.SQL;
+import org.springframework.data.relational.core.sql.Table;
+
+/**
+ * Utility to get from path to SQL DSL elements.
+ *
+ * @author Jens Schauder
+ * @since 1.1
+ */
+class SqlContext {
+
+	private final RelationalPersistentEntity<?> entity;
+
+	SqlContext(RelationalPersistentEntity<?> entity) {
+		this.entity = entity;
+	}
+
+	Column getIdColumn() {
+		return getTable().column(entity.getIdColumn());
+	}
+
+	Table getTable(PersistentPropertyPathExtension path) {
+
+		String tableAlias = path.getTableAlias();
+		Table table = SQL.table(path.getTableName());
+		return tableAlias == null ? table : table.as(tableAlias);
+	}
+
+	Table getTable() {
+		return SQL.table(entity.getTableName());
+	}
+
+	Column getColumn(PersistentPropertyPathExtension path) {
+		return getTable(path).column(path.getColumnName()).as(path.getColumnAlias());
+	}
+
+	Column getReverseColumn(PersistentPropertyPathExtension path) {
+		return getTable(path).column(path.getReverseColumnName()).as(path.getReverseColumnNameAlias());
+	}
+}

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

@@ -15,13 +15,15 @@
  */
 package org.springframework.data.jdbc.core;
 
+import lombok.Value;
+
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -33,6 +35,8 @@ import org.springframework.data.mapping.PropertyHandler;
 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.relational.core.sql.*;
+import org.springframework.data.relational.core.sql.render.SqlRenderer;
 import org.springframework.data.util.Lazy;
 import org.springframework.data.util.StreamUtils;
 import org.springframework.lang.Nullable;
@@ -49,12 +53,12 @@ import org.springframework.util.Assert;
 class SqlGenerator {
 
 	private final RelationalPersistentEntity<?> entity;
-	private final RelationalMappingContext context;
+	private final RelationalMappingContext mappingContext;
 	private final List<String> columnNames = new ArrayList<>();
 	private final List<String> nonIdColumnNames = new ArrayList<>();
 	private final Set<String> readOnlyColumnNames = new HashSet<>();
 
-	private final Lazy<String> findOneSql = Lazy.of(this::createFindOneSelectSql);
+	private final Lazy<String> findOneSql = Lazy.of(this::createFindOneSql);
 	private final Lazy<String> findAllSql = Lazy.of(this::createFindAllSql);
 	private final Lazy<String> findAllInListSql = Lazy.of(this::createFindAllInListSql);
 
@@ -68,13 +72,15 @@ class SqlGenerator {
 	private final SqlGeneratorSource sqlGeneratorSource;
 
 	private final Pattern parameterPattern = Pattern.compile("\\W");
+	private final SqlContext sqlContext;
 
-	SqlGenerator(RelationalMappingContext context, RelationalPersistentEntity<?> entity,
+	SqlGenerator(RelationalMappingContext mappingContext, RelationalPersistentEntity<?> entity,
 			SqlGeneratorSource sqlGeneratorSource) {
 
-		this.context = context;
+		this.mappingContext = mappingContext;
 		this.entity = entity;
 		this.sqlGeneratorSource = sqlGeneratorSource;
+		this.sqlContext = new SqlContext(entity);
 		initColumnNames(entity, "");
 	}
 
@@ -109,7 +115,8 @@ class SqlGenerator {
 
 		final String embeddedPrefix = property.getEmbeddedPrefix();
 
-		final RelationalPersistentEntity<?> embeddedEntity = context.getRequiredPersistentEntity(property.getColumnType());
+		final RelationalPersistentEntity<?> embeddedEntity = mappingContext
+				.getRequiredPersistentEntity(property.getColumnType());
 
 		initColumnNames(embeddedEntity, prefix + embeddedPrefix);
 	}
@@ -150,14 +157,20 @@ class SqlGenerator {
 		Assert.isTrue(keyColumn != null || !ordered,
 				"If the SQL statement should be ordered a keyColumn to order by must be provided.");
 
-		String baseSelect = (keyColumn != null) //
-				? createSelectBuilder().column(cb -> cb.tableAlias(entity.getTableName()).column(keyColumn).as(keyColumn))
-						.build()
-				: getFindAll();
+		SelectBuilder.SelectWhere baseSelect = createBaseSelect(keyColumn);
+
+		Table table = Table.create(entity.getTableName());
+		SelectBuilder.SelectWhereAndOr withWhereClause = baseSelect
+				.where(table.column(columnName).isEqualTo(SQL.bindMarker(":" + columnName)));
 
-		String orderBy = ordered ? " ORDER BY " + keyColumn : "";
+		SelectBuilder.BuildSelect select;
+		if (ordered) {
+			select = withWhereClause.orderBy(table.column(keyColumn).as(keyColumn));
+		} else {
+			select = withWhereClause;
+		}
 
-		return String.format("%s WHERE %s = :%s%s", baseSelect, columnName, columnName, orderBy);
+		return render(select);
 	}
 
 	String getExists() {
@@ -188,277 +201,311 @@ class SqlGenerator {
 		return deleteByListSql.get();
 	}
 
-	private String createFindOneSelectSql() {
+	private String createFindOneSql() {
 
-		return createSelectBuilder() //
-				.where(wb -> wb.tableAlias(entity.getTableName()).column(entity.getIdColumn()).eq().variable("id")) //
-				.build();
+		SelectBuilder.SelectWhereAndOr withCondition = createBaseSelect()
+				.where(sqlContext.getIdColumn().isEqualTo(SQL.bindMarker(":id")));
+
+		return render(withCondition);
 	}
 
-	private SelectBuilder createSelectBuilder() {
+	private Stream<String> getColumnNameStream(String prefix) {
 
-		SelectBuilder builder = new SelectBuilder(entity.getTableName());
-		addColumnsForSimpleProperties(entity, "", "", entity, builder);
-		addColumnsForEmbeddedProperties(entity, "", "", entity, builder);
-		addColumnsAndJoinsForOneToOneReferences(entity, "", "", entity, builder);
+		return StreamUtils.createStreamFromIterator(entity.iterator()) //
+				.flatMap(p -> getColumnNameStream(p, prefix));
+	}
+
+	private Stream<String> getColumnNameStream(RelationalPersistentProperty p, String prefix) {
 
-		return builder;
+		if (p.isEntity()) {
+			return sqlGeneratorSource.getSqlGenerator(p.getType()).getColumnNameStream(prefix + p.getColumnName() + "_");
+		} else {
+			return Stream.of(prefix + p.getColumnName());
+		}
 	}
 
-	/**
-	 * Adds the columns to the provided {@link SelectBuilder} representing simple properties, including those from
-	 * one-to-one relationships.
-	 *
-	 * @param rootEntity the root entity for which to add the columns.
-	 * @param builder The {@link SelectBuilder} to be modified.
-	 */
-	private void addColumnsAndJoinsForOneToOneReferences(RelationalPersistentEntity<?> entity, String prefix,
-			String tableAlias, RelationalPersistentEntity<?> rootEntity, SelectBuilder builder) {
-
-		for (RelationalPersistentProperty property : entity) {
-			if (!property.isEntity() //
-					|| property.isEmbedded() //
-					|| Collection.class.isAssignableFrom(property.getType()) //
-					|| Map.class.isAssignableFrom(property.getType()) //
-			) {
-				continue;
-			}
+	private String createFindAllSql() {
+		return render(createBaseSelect());
+	}
 
-			final RelationalPersistentEntity<?> refEntity = context.getRequiredPersistentEntity(property.getActualType());
-			final String joinAlias;
-
-			if (tableAlias.isEmpty()) {
-				if (prefix.isEmpty()) {
-					joinAlias = property.getName();
-				} else {
-					joinAlias = prefix + property.getName();
-				}
-			} else {
-				if (prefix.isEmpty()) {
-					joinAlias = tableAlias + "_" + property.getName();
-				} else {
-					joinAlias = tableAlias + "_" + prefix + property.getName();
-				}
-			}
+	private SelectBuilder.SelectWhere createBaseSelect() {
 
-			// final String joinAlias = tableAlias.isEmpty() ? property.getName() : tableAlias + "_" + property.getName();
-			builder.join(jb -> jb.leftOuter().table(refEntity.getTableName()).as(joinAlias) //
-					.where(property.getReverseColumnName()).eq().column(rootEntity.getTableName(), rootEntity.getIdColumn()));
+		return createBaseSelect(null);
+	}
 
-			addColumnsForSimpleProperties(refEntity, "", joinAlias, refEntity, builder);
-			addColumnsForEmbeddedProperties(refEntity, "", joinAlias, refEntity, builder);
-			addColumnsAndJoinsForOneToOneReferences(refEntity, "", joinAlias, refEntity, builder);
+	private SelectBuilder.SelectWhere createBaseSelect(@Nullable String keyColumn) {
 
-			// if the referenced property doesn't have an id, include the back reference in the select list.
-			// this enables determining if the referenced entity is present or null.
-			if (!refEntity.hasIdProperty()) {
+		Table table = SQL.table(entity.getTableName());
 
-				builder.column( //
-						cb -> cb.tableAlias(joinAlias) //
-								.column(property.getReverseColumnName()) //
-								.as(joinAlias + "_" + property.getReverseColumnName()) //
-				);
+		List<Expression> columnExpressions = new ArrayList<>();
+
+		List<Join> joinTables = new ArrayList<>();
+		for (PersistentPropertyPath<RelationalPersistentProperty> path : mappingContext
+				.findPersistentPropertyPaths(entity.getType(), p -> true)) {
+
+			PersistentPropertyPathExtension extPath = new PersistentPropertyPathExtension(mappingContext, path);
+
+			// add a join if necessary
+			Join join = getJoin(extPath);
+			if (join != null) {
+				joinTables.add(join);
 			}
-		}
-	}
 
-	private void addColumnsForEmbeddedProperties(RelationalPersistentEntity<?> currentEntity, String prefix,
-			String tableAlias, RelationalPersistentEntity<?> rootEntity, SelectBuilder builder) {
-		for (RelationalPersistentProperty property : currentEntity) {
-			if (!property.isEmbedded()) {
-				continue;
+			Column column = getColumn(extPath);
+			if (column != null) {
+				columnExpressions.add(column);
 			}
+		}
 
-			final String embeddedPrefix = prefix + property.getEmbeddedPrefix();
-			final RelationalPersistentEntity<?> embeddedEntity = context
-					.getRequiredPersistentEntity(property.getColumnType());
+		if (keyColumn != null) {
+			columnExpressions.add(table.column(keyColumn).as(keyColumn));
+		}
+
+		SelectBuilder.SelectAndFrom selectBuilder = StatementBuilder.select(columnExpressions);
+
+		SelectBuilder.SelectJoin baseSelect = selectBuilder.from(table);
 
-			addColumnsForSimpleProperties(embeddedEntity, embeddedPrefix, tableAlias, rootEntity, builder);
-			addColumnsForEmbeddedProperties(embeddedEntity, embeddedPrefix, tableAlias, rootEntity, builder);
-			addColumnsAndJoinsForOneToOneReferences(embeddedEntity, embeddedPrefix, tableAlias, rootEntity, builder);
+		for (Join join : joinTables) {
+			baseSelect = baseSelect.leftOuterJoin(join.joinTable).on(join.joinColumn).equals(join.parentId);
 		}
+
+		return (SelectBuilder.SelectWhere) baseSelect;
 	}
 
-	private void addColumnsForSimpleProperties(RelationalPersistentEntity<?> currentEntity, String prefix,
-			String tableAlias, RelationalPersistentEntity<?> rootEntity, SelectBuilder builder) {
+	@Nullable
+	Column getColumn(PersistentPropertyPathExtension path) {
+
+		// an embedded itself doesn't give an column, its members will though.
+		// if there is a collection or map on the path it won't get selected at all, but it will get loaded with a separate
+		// select
+		// only the parent path is considered in order to handle arrays that get stored as BINARY properly
+		if (path.isEmbedded() || path.getParentPath().isMultiValued()) {
+			return null;
+		}
+
+		if (path.isEntity()) {
 
-		for (RelationalPersistentProperty property : currentEntity) {
+			// Simple entities without id include there backreference as an synthetic id in order to distinguish null entities
+			// from entities with only null values.
 
-			if (property.isEntity()) {
-				continue;
+			if (path.isQualified() //
+					|| path.isCollectionLike() //
+					|| path.hasIdProperty() //
+			) {
+				return null;
 			}
 
-			final String column = prefix + property.getColumnName();
-			final String as = tableAlias.isEmpty() ? column : tableAlias + "_" + column;
+			return sqlContext.getReverseColumn(path);
 
-			builder.column(cb -> cb //
-					.tableAlias(tableAlias.isEmpty() ? rootEntity.getTableName() : tableAlias) //
-					.column(column) //
-					.as(as));
 		}
-	}
 
-	private Stream<String> getColumnNameStream(String prefix) {
+		return sqlContext.getColumn(path);
 
-		return StreamUtils.createStreamFromIterator(entity.iterator()) //
-				.flatMap(p -> getColumnNameStream(p, prefix));
 	}
 
-	private Stream<String> getColumnNameStream(RelationalPersistentProperty p, String prefix) {
+	@Nullable
+	Join getJoin(PersistentPropertyPathExtension path) {
 
-		if (p.isEntity()) {
-			return sqlGeneratorSource.getSqlGenerator(p.getType()).getColumnNameStream(prefix + p.getColumnName() + "_");
-		} else {
-			return Stream.of(prefix + p.getColumnName());
+		if (!path.isEntity() || path.isEmbedded() || path.isMultiValued()) {
+			return null;
 		}
-	}
 
-	private String createFindAllSql() {
-		return createSelectBuilder().build();
+		Table currentTable = sqlContext.getTable(path);
+
+		PersistentPropertyPathExtension idDefiningParentPath = path.getIdDefiningParentPath();
+		Table parentTable = sqlContext.getTable(idDefiningParentPath);
+
+		return new Join( //
+				currentTable, //
+				currentTable.column(path.getReverseColumnName()), //
+				parentTable.column(idDefiningParentPath.getIdColumnName()) //
+		);
 	}
 
 	private String createFindAllInListSql() {
 
-		return createSelectBuilder() //
-				.where(wb -> wb.tableAlias(entity.getTableName()).column(entity.getIdColumn()).in().variable("ids")) //
-				.build();
+		SelectBuilder.SelectWhereAndOr withCondition = createBaseSelect()
+				.where(sqlContext.getIdColumn().in(SQL.bindMarker(":ids")));
+
+		return render(withCondition);
+	}
+
+	private String render(SelectBuilder.BuildSelect select) {
+		return SqlRenderer.create().render(select.build());
+	}
+
+	private String render(InsertBuilder.BuildInsert insert) {
+		return SqlRenderer.create().render(insert.build());
+	}
+
+	private String render(DeleteBuilder.BuildDelete delete) {
+		return SqlRenderer.create().render(delete.build());
+	}
+
+	private String render(UpdateBuilder.BuildUpdate update) {
+		return SqlRenderer.create().render(update.build());
 	}
 
 	private String createExistsSql() {
-		return String.format("SELECT COUNT(*) FROM %s WHERE %s = :id", entity.getTableName(), entity.getIdColumn());
+
+		Table table = sqlContext.getTable();
+		Column idColumn = table.column(entity.getIdColumn());
+
+		SelectBuilder.BuildSelect select = StatementBuilder //
+				.select(Functions.count(idColumn)) //
+				.from(table) //
+				.where(idColumn.isEqualTo(SQL.bindMarker(":id")));
+
+		return render(select);
 	}
 
 	private String createCountSql() {
-		return String.format("select count(*) from %s", entity.getTableName());
+
+		Table table = SQL.table(entity.getTableName());
+
+		SelectBuilder.BuildSelect select = StatementBuilder //
+				.select(Functions.count(Expressions.asterisk())) //
+				.from(table);
+
+		return render(select);
 	}
 
 	private String createInsertSql(Set<String> additionalColumns) {
 
-		String insertTemplate = "INSERT INTO %s (%s) VALUES (%s)";
+		Table table = SQL.table(entity.getTableName());
 
 		LinkedHashSet<String> columnNamesForInsert = new LinkedHashSet<>(nonIdColumnNames);
 		columnNamesForInsert.addAll(additionalColumns);
 		columnNamesForInsert.removeIf(readOnlyColumnNames::contains);
 
-		String tableColumns = String.join(", ", columnNamesForInsert);
+		InsertBuilder.InsertIntoColumnsAndValuesWithBuild insert = Insert.builder().into(table);
 
-		String parameterNames = columnNamesForInsert.stream()//
-				.map(this::columnNameToParameterName)
-				.map(n -> String.format(":%s", n))//
-				.collect(Collectors.joining(", "));
+		for (String cn : columnNamesForInsert) {
+			insert = insert.column(table.column(cn));
+		}
+
+		InsertBuilder.InsertValuesWithBuild insertWithValues = null;
+		for (String cn : columnNamesForInsert) {
+			insertWithValues = (insertWithValues == null ? insert : insertWithValues)
+					.values(SQL.bindMarker(":" + columnNameToParameterName(cn)));
+		}
 
-		return String.format(insertTemplate, entity.getTableName(), tableColumns, parameterNames);
+		return render(insertWithValues == null ? insert : insertWithValues);
 	}
 
 	private String createUpdateSql() {
 
-		String updateTemplate = "UPDATE %s SET %s WHERE %s = :%s";
+		Table table = SQL.table(entity.getTableName());
 
-		String setClause = columnNames.stream() //
+		List<AssignValue> assignments = columnNames.stream() //
 				.filter(s -> !s.equals(entity.getIdColumn())) //
 				.filter(s -> !readOnlyColumnNames.contains(s)) //
-				.map(n -> String.format("%s = :%s", n, columnNameToParameterName(n))) //
-				.collect(Collectors.joining(", "));
-
-		return String.format( //
-				updateTemplate, //
-				entity.getTableName(), //
-				setClause, //
-				entity.getIdColumn(), //
-				columnNameToParameterName(entity.getIdColumn()) //
-		);
+				.map(columnName -> Assignments.value( //
+						table.column(columnName), //
+						SQL.bindMarker(":" + columnNameToParameterName(columnName)))) //
+				.collect(Collectors.toList());
+
+		UpdateBuilder.UpdateWhereAndOr update = Update.builder() //
+				.table(table) //
+				.set(assignments) //
+				.where(table.column(entity.getIdColumn())
+						.isEqualTo(SQL.bindMarker(":" + columnNameToParameterName(entity.getIdColumn())))) //
+		;
+
+		return render(update);
 	}
 
 	private String createDeleteSql() {
-		return String.format("DELETE FROM %s WHERE %s = :id", entity.getTableName(), entity.getIdColumn());
+
+		Table table = SQL.table(entity.getTableName());
+
+		DeleteBuilder.DeleteWhereAndOr delete = Delete.builder().from(table)
+				.where(table.column(entity.getIdColumn()).isEqualTo(SQL.bindMarker(":id")));
+
+		return render(delete);
 	}
 
 	String createDeleteAllSql(@Nullable PersistentPropertyPath<RelationalPersistentProperty> path) {
 
-		if (path == null) {
-			return String.format("DELETE FROM %s", entity.getTableName());
-		}
-
-		RelationalPersistentEntity<?> entityToDelete = context
-				.getRequiredPersistentEntity(path.getRequiredLeafProperty().getActualType());
+		Table table = SQL.table(entity.getTableName());
 
-		final String innerMostCondition1 = createInnerMostCondition("%s IS NOT NULL", path);
-		String condition = cascadeConditions(innerMostCondition1, getSubPath(path));
+		DeleteBuilder.DeleteWhere deleteAll = Delete.builder().from(table);
 
-		return String.format("DELETE FROM %s WHERE %s", entityToDelete.getTableName(), condition);
+		if (path == null) {
+			return render(deleteAll);
+		}
+		return createDeleteByPathAndCriteria(new PersistentPropertyPathExtension(mappingContext, path), Column::isNotNull);
 	}
 
 	private String createDeleteByListSql() {
-		return String.format("DELETE FROM %s WHERE %s IN (:ids)", entity.getTableName(), entity.getIdColumn());
-	}
 
-	String createDeleteByPath(PersistentPropertyPath<RelationalPersistentProperty> path) {
+		Table table = SQL.table(entity.getTableName());
 
-		RelationalPersistentEntity<?> entityToDelete = context
-				.getRequiredPersistentEntity(path.getRequiredLeafProperty().getActualType());
+		DeleteBuilder.DeleteWhereAndOr delete = Delete.builder() //
+				.from(table) //
+				.where(table.column(entity.getIdColumn()).in(SQL.bindMarker(":ids")));
 
-		final String innerMostCondition = createInnerMostCondition("%s = :rootId", path);
-		String condition = cascadeConditions(innerMostCondition, getSubPath(path));
+		return render(delete);
+	}
 
-		return String.format("DELETE FROM %s WHERE %s", entityToDelete.getTableName(), condition);
+	String createDeleteByPath(PersistentPropertyPath<RelationalPersistentProperty> path) {
+		return createDeleteByPathAndCriteria(new PersistentPropertyPathExtension(mappingContext, path),
+				filterColumn -> filterColumn.isEqualTo(SQL.bindMarker(":rootId")));
 	}
 
-	private String createInnerMostCondition(String template, PersistentPropertyPath<RelationalPersistentProperty> path) {
-		PersistentPropertyPath<RelationalPersistentProperty> currentPath = path;
-		while (!currentPath.getParentPath().isEmpty()
-				&& !currentPath.getParentPath().getRequiredLeafProperty().isEmbedded()) {
-			currentPath = currentPath.getParentPath();
-		}
+	private String createDeleteByPathAndCriteria(PersistentPropertyPathExtension path,
+			Function<Column, Condition> rootCondition) {
 
-		RelationalPersistentProperty property = currentPath.getRequiredLeafProperty();
-		return String.format(template, property.getReverseColumnName());
-	}
+		Table table = SQL.table(path.getTableName());
 
-	private PersistentPropertyPath<RelationalPersistentProperty> getSubPath(
-			PersistentPropertyPath<RelationalPersistentProperty> path) {
+		DeleteBuilder.DeleteWhere delete = Delete.builder() //
+				.from(table);
 
-		int pathLength = path.getLength();
+		DeleteBuilder.DeleteWhereAndOr deleteWithWhere;
 
-		PersistentPropertyPath<RelationalPersistentProperty> ancestor = path;
+		Column filterColumn = table.column(path.getReverseColumnName());
 
-		int embeddedDepth = 0;
-		while (!ancestor.getParentPath().isEmpty() && ancestor.getParentPath().getRequiredLeafProperty().isEmbedded()) {
-			embeddedDepth++;
-			ancestor = ancestor.getParentPath();
-		}
+		if (path.getLength() == 1) {
 
-		ancestor = path;
+			deleteWithWhere = delete //
+					.where(rootCondition.apply(filterColumn));
+		} else {
 
-		for (int i = pathLength - 1 + embeddedDepth; i > 0; i--) {
-			ancestor = ancestor.getParentPath();
+			Condition condition = getSubselectCondition(path, rootCondition, filterColumn);
+			deleteWithWhere = delete.where(condition);
 		}
-
-		return path.getExtensionForBaseOf(ancestor);
+		return render(deleteWithWhere);
 	}
 
-	private String cascadeConditions(String innerCondition, PersistentPropertyPath<RelationalPersistentProperty> path) {
+	private Condition getSubselectCondition(PersistentPropertyPathExtension path,
+			Function<Column, Condition> rootCondition, Column filterColumn) {
 
-		if (path.getLength() == 0) {
-			return innerCondition;
-		}
+		PersistentPropertyPathExtension parentPath = path.getParentPath();
 
-		PersistentPropertyPath<RelationalPersistentProperty> rootPath = path;
-		while (rootPath.getLength() > 1) {
-			rootPath = rootPath.getParentPath();
-		}
+		Table subSelectTable = SQL.table(parentPath.getTableName());
+		Column idColumn = subSelectTable.column(parentPath.getIdColumnName());
+		Column selectFilterColumn = subSelectTable.column(parentPath.getEffectiveIdColumnName());
 
-		RelationalPersistentEntity<?> entity = context
-				.getRequiredPersistentEntity(rootPath.getBaseProperty().getOwner().getTypeInformation());
-		RelationalPersistentProperty property = path.getRequiredLeafProperty();
+		Condition innerCondition = parentPath.getLength() == 1 ? rootCondition.apply(selectFilterColumn)
+				: getSubselectCondition(parentPath, rootCondition, selectFilterColumn);
 
-		return String.format("%s IN (SELECT %s FROM %s WHERE %s)", //
-				property.getReverseColumnName(), //
-				entity.getIdColumn(), //
-				entity.getTableName(), innerCondition //
-		);
+		Select select = Select.builder() //
+				.select(idColumn) //
+				.from(subSelectTable) //
+				.where(innerCondition).build();
+
+		return filterColumn.in(select);
 	}
 
-	private String columnNameToParameterName(String columnName){
+	private String columnNameToParameterName(String columnName) {
 		return parameterPattern.matcher(columnName).replaceAll("");
 	}
+
+	@Value
+	class Join {
+		Table joinTable;
+		Column joinColumn;
+		Column parentId;
+	}
+
 }

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

@@ -21,7 +21,6 @@ import static org.mockito.Mockito.*;
 
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
-
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
 import org.springframework.data.relational.core.conversion.DbAction.Insert;
@@ -55,7 +54,7 @@ public class DefaultJdbcInterpreterUnitTests {
 	Element element = new Element();
 
 	InsertRoot<Container> containerInsert = new InsertRoot<>(container);
-	Insert<?> insert = new Insert<>(element, PropertyPathUtils.toPath("element", Container.class, context),
+	Insert<?> insert = new Insert<>(element, PropertyPathTestingUtils.toPath("element", Container.class, context),
 			containerInsert);
 
 	@Test // DATAJDBC-145
@@ -104,6 +103,7 @@ public class DefaultJdbcInterpreterUnitTests {
 				.containsExactly(tuple(BACK_REFERENCE, CONTAINER_ID, Long.class));
 	}
 
+	@SuppressWarnings("unused")
 	static class Container {
 
 		@Id Long id;

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

@@ -20,6 +20,8 @@ import static org.assertj.core.api.Assertions.*;
 
 import lombok.Data;
 
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -31,7 +33,6 @@ import org.junit.Assume;
 import org.junit.ClassRule;
 import org.junit.Rule;
 import org.junit.Test;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.annotation.Bean;
@@ -44,6 +45,9 @@ import org.springframework.data.relational.core.conversion.RelationalConverter;
 import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.Table;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.annotation.ProfileValueSourceConfiguration;
 import org.springframework.test.annotation.ProfileValueUtils;
@@ -67,7 +71,8 @@ public class JdbcAggregateTemplateIntegrationTests {
 	@ClassRule public static final SpringClassRule classRule = new SpringClassRule();
 	@Rule public SpringMethodRule methodRule = new SpringMethodRule();
 	@Autowired JdbcAggregateOperations template;
-
+	@Autowired
+	NamedParameterJdbcOperations jdbcTemplate;
 	LegoSet legoSet = createLegoSet();
 
 	@Test // DATAJDBC-112
@@ -422,6 +427,7 @@ public class JdbcAggregateTemplateIntegrationTests {
 
 	@Test // DATAJDBC-327
 	public void saveAndLoadAnEntityWithByteArray() {
+
 		ByteArrayOwner owner = new ByteArrayOwner();
 		owner.binaryData = new byte[] { 1, 23, 42 };
 
@@ -434,6 +440,35 @@ public class JdbcAggregateTemplateIntegrationTests {
 		assertThat(reloaded.binaryData).isEqualTo(new byte[] { 1, 23, 42 });
 	}
 
+	@Test
+	public void saveAndLoadLongChain() {
+
+		Chain4 chain4 = new Chain4();
+		chain4.fourValue = "omega";
+		chain4.chain3 = new Chain3();
+		chain4.chain3.threeValue = "delta";
+		chain4.chain3.chain2 = new Chain2();
+		chain4.chain3.chain2.twoValue = "gamma";
+		chain4.chain3.chain2.chain1 = new Chain1();
+		chain4.chain3.chain2.chain1.oneValue = "beta";
+		chain4.chain3.chain2.chain1.chain0 = new Chain0();
+		chain4.chain3.chain2.chain1.chain0.zeroValue = "alpha";
+
+		template.save(chain4);
+
+		Chain4 reloaded = template.findById(chain4.four, Chain4.class);
+
+		assertThat(reloaded).isNotNull();
+
+		assertThat(reloaded.four).isEqualTo(chain4.four);
+		assertThat(reloaded.chain3.chain2.chain1.chain0.zeroValue).isEqualTo(chain4.chain3.chain2.chain1.chain0.zeroValue);
+
+		template.delete(chain4, Chain4.class);
+
+		String countSelect = "SELECT COUNT(*) FROM %s";
+		jdbcTemplate.queryForObject(String.format(countSelect, "CHAIN0"),emptyMap(), Long.class);
+	}
+
 	private static void assumeNot(String dbProfileName) {
 
 		Assume.assumeTrue("true"
@@ -537,4 +572,33 @@ public class JdbcAggregateTemplateIntegrationTests {
 			return new JdbcAggregateTemplate(publisher, context, converter, dataAccessStrategy);
 		}
 	}
+
+	static class Chain0 {
+		@Id Long zero;
+		String zeroValue;
+	}
+
+	static class Chain1 {
+		@Id Long one;
+		String oneValue;
+		Chain0 chain0;
+	}
+
+	static class Chain2 {
+		@Id Long two;
+		String twoValue;
+		Chain1 chain1;
+	}
+
+	static class Chain3 {
+		@Id Long three;
+		String threeValue;
+		Chain2 chain2;
+	}
+
+	static class Chain4 {
+		@Id Long four;
+		String fourValue;
+		Chain3 chain3;
+	}
 }

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

@@ -16,7 +16,7 @@
 package org.springframework.data.jdbc.core;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.springframework.data.jdbc.core.PropertyPathUtils.*;
+import static org.springframework.data.jdbc.core.PropertyPathTestingUtils.*;
 
 import java.util.List;
 import java.util.Map;
@@ -24,7 +24,6 @@ import java.util.UUID;
 
 import org.jetbrains.annotations.NotNull;
 import org.junit.Test;
-
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.convert.JdbcIdentifierBuilder;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
@@ -49,8 +48,8 @@ public class JdbcIdentifierBuilderUnitTests {
 		assertThat(identifier.getParts()) //
 				.extracting("name", "value", "targetType") //
 				.containsExactly( //
-				tuple("dummy_entity", "eins", UUID.class) //
-		);
+						tuple("dummy_entity", "eins", UUID.class) //
+				);
 	}
 
 	@Test // DATAJDBC-326
@@ -66,9 +65,9 @@ public class JdbcIdentifierBuilderUnitTests {
 		assertThat(identifier.getParts()) //
 				.extracting("name", "value", "targetType") //
 				.containsExactlyInAnyOrder( //
-				tuple("dummy_entity", "parent-eins", UUID.class), //
-				tuple("dummy_entity_key", "map-key-eins", String.class) //
-		);
+						tuple("dummy_entity", "parent-eins", UUID.class), //
+						tuple("dummy_entity_key", "map-key-eins", String.class) //
+				);
 	}
 
 	@Test // DATAJDBC-326
@@ -84,9 +83,9 @@ public class JdbcIdentifierBuilderUnitTests {
 		assertThat(identifier.getParts()) //
 				.extracting("name", "value", "targetType") //
 				.containsExactlyInAnyOrder( //
-				tuple("dummy_entity", "parent-eins", UUID.class), //
-				tuple("dummy_entity_key", "list-index-eins", Integer.class) //
-		);
+						tuple("dummy_entity", "parent-eins", UUID.class), //
+						tuple("dummy_entity_key", "list-index-eins", Integer.class) //
+				);
 	}
 
 	@Test // DATAJDBC-326
@@ -99,8 +98,8 @@ public class JdbcIdentifierBuilderUnitTests {
 		assertThat(identifier.getParts()) //
 				.extracting("name", "value", "targetType") //
 				.containsExactly( //
-				tuple("embeddable", "parent-eins", UUID.class) //
-		);
+						tuple("embeddable", "parent-eins", UUID.class) //
+				);
 	}
 
 	@NotNull

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

@@ -47,11 +47,7 @@ public class MyBatisDataAccessStrategyUnitTests {
 
 	MyBatisDataAccessStrategy accessStrategy = new MyBatisDataAccessStrategy(session);
 
-	PersistentPropertyPath<RelationalPersistentProperty> path(String path, Class source) {
-
-		RelationalMappingContext context = this.context;
-		return PropertyPathUtils.toPath(path, source, context);
-	}
+	PersistentPropertyPath<RelationalPersistentProperty> path = PropertyPathTestingUtils.toPath("one.two", DummyEntity.class, context);
 
 	@Before
 	public void before() {
@@ -128,7 +124,7 @@ public class MyBatisDataAccessStrategyUnitTests {
 	@Test // DATAJDBC-123
 	public void deleteAllByPath() {
 
-		accessStrategy.deleteAll(path("one.two", DummyEntity.class));
+		accessStrategy.deleteAll(path);
 
 		verify(session).delete(
 				eq("org.springframework.data.jdbc.core.MyBatisDataAccessStrategyUnitTests$DummyEntityMapper.deleteAll-one-two"),
@@ -174,7 +170,7 @@ public class MyBatisDataAccessStrategyUnitTests {
 	@Test // DATAJDBC-123
 	public void deleteByPath() {
 
-		accessStrategy.delete("rootid", path("one.two", DummyEntity.class));
+		accessStrategy.delete("rootid", path);
 
 		verify(session).delete(
 				eq("org.springframework.data.jdbc.core.MyBatisDataAccessStrategyUnitTests$DummyEntityMapper.delete-one-two"),
@@ -334,10 +330,12 @@ public class MyBatisDataAccessStrategyUnitTests {
 				);
 	}
 
+	@SuppressWarnings("unused")
 	private static class DummyEntity {
 		ChildOne one;
 	}
 
+	@SuppressWarnings("unused")
 	private static class ChildOne {
 		ChildTwo two;
 	}

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

@@ -0,0 +1,174 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://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;
+
+import java.util.List;
+
+import org.assertj.core.api.SoftAssertions;
+import org.jetbrains.annotations.NotNull;
+import org.junit.Test;
+import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.relational.core.mapping.Embedded;
+import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+
+/**
+ * @author Jens Schauder
+ */
+public class PersistentPropertyPathExtensionUnitTests {
+
+	JdbcMappingContext context = new JdbcMappingContext();
+	private RelationalPersistentEntity<?> entity = context.getRequiredPersistentEntity(DummyEntity.class);
+
+	@Test
+	public void isEmbedded() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath(entity).isEmbedded()).isFalse();
+			softly.assertThat(extPath("second").isEmbedded()).isFalse();
+			softly.assertThat(extPath("second.third").isEmbedded()).isTrue();
+		});
+	}
+
+	@Test
+	public void isMultiValued() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath(entity).isMultiValued()).isFalse();
+			softly.assertThat(extPath("second").isMultiValued()).isFalse();
+			softly.assertThat(extPath("second.third").isMultiValued()).isFalse();
+			softly.assertThat(extPath("secondList.third").isMultiValued()).isTrue();
+			softly.assertThat(extPath("secondList").isMultiValued()).isTrue();
+		});
+	}
+
+	@Test
+	public void leafEntity() {
+
+		RelationalPersistentEntity<?> second = context.getRequiredPersistentEntity(Second.class);
+		RelationalPersistentEntity<?> third = context.getRequiredPersistentEntity(Third.class);
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath(entity).getLeafEntity()).isEqualTo(entity);
+			softly.assertThat(extPath("second").getLeafEntity()).isEqualTo(second);
+			softly.assertThat(extPath("second.third").getLeafEntity()).isEqualTo(third);
+			softly.assertThat(extPath("secondList.third").getLeafEntity()).isEqualTo(third);
+			softly.assertThat(extPath("secondList").getLeafEntity()).isEqualTo(second);
+		});
+	}
+
+	@Test
+	public void isEntity() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath(entity).isEntity()).isTrue();
+			String path = "second";
+			softly.assertThat(extPath(path).isEntity()).isTrue();
+			softly.assertThat(extPath("second.third").isEntity()).isTrue();
+			softly.assertThat(extPath("second.third.value").isEntity()).isFalse();
+			softly.assertThat(extPath("secondList.third").isEntity()).isTrue();
+			softly.assertThat(extPath("secondList.third.value").isEntity()).isFalse();
+			softly.assertThat(extPath("secondList").isEntity()).isTrue();
+		});
+	}
+
+	@Test
+	public void getTableName() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath(entity).getTableName()).isEqualTo("dummy_entity");
+			softly.assertThat(extPath("second").getTableName()).isEqualTo("second");
+			softly.assertThat(extPath("second.third").getTableName()).isEqualTo("second");
+			softly.assertThat(extPath("second.third.value").getTableName()).isEqualTo("second");
+			softly.assertThat(extPath("secondList.third").getTableName()).isEqualTo("second");
+			softly.assertThat(extPath("secondList.third.value").getTableName()).isEqualTo("second");
+			softly.assertThat(extPath("secondList").getTableName()).isEqualTo("second");
+		});
+	}
+
+	@Test
+	public void getTableAlias() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath(entity).getTableAlias()).isEqualTo(null);
+			softly.assertThat(extPath("second").getTableAlias()).isEqualTo("second");
+			softly.assertThat(extPath("second.third").getTableAlias()).isEqualTo("second");
+			softly.assertThat(extPath("second.third.value").getTableAlias()).isEqualTo("second");
+			softly.assertThat(extPath("second.third2").getTableAlias()).isEqualTo("second_third2");
+			softly.assertThat(extPath("second.third2.value").getTableAlias()).isEqualTo("second_third2");
+			softly.assertThat(extPath("secondList.third").getTableAlias()).isEqualTo("secondList");
+			softly.assertThat(extPath("secondList.third.value").getTableAlias()).isEqualTo("secondList");
+			softly.assertThat(extPath("secondList.third2").getTableAlias()).isEqualTo("secondList_third2");
+			softly.assertThat(extPath("secondList.third2.value").getTableAlias()).isEqualTo("secondList_third2");
+			softly.assertThat(extPath("secondList").getTableAlias()).isEqualTo("secondList");
+			softly.assertThat(extPath("second2.third2").getTableAlias()).isEqualTo("secthird2");
+		});
+	}
+
+	@Test
+	public void getColumnName() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(extPath("second.third.value").getColumnName()).isEqualTo("thrdvalue");
+			softly.assertThat(extPath("second.third2.value").getColumnName()).isEqualTo("value");
+			softly.assertThat(extPath("secondList.third.value").getColumnName()).isEqualTo("thrdvalue");
+			softly.assertThat(extPath("secondList.third2.value").getColumnName()).isEqualTo("value");
+			softly.assertThat(extPath("second2.third.value").getColumnName()).isEqualTo("secthrdvalue");
+			softly.assertThat(extPath("second2.third2.value").getColumnName()).isEqualTo("value");
+		});
+	}
+
+	@NotNull
+	private PersistentPropertyPathExtension extPath(RelationalPersistentEntity<?> entity) {
+		return new PersistentPropertyPathExtension(context, entity);
+	}
+
+	@NotNull
+	private PersistentPropertyPathExtension extPath(String path) {
+		return new PersistentPropertyPathExtension(context, createSimplePath(path));
+	}
+
+	PersistentPropertyPath<RelationalPersistentProperty> createSimplePath(String path) {
+		return PropertyPathTestingUtils.toPath(path, DummyEntity.class, context);
+	}
+
+	@SuppressWarnings("unused")
+	static class DummyEntity {
+		Second second;
+		List<Second> secondList;
+		@Embedded("sec") Second second2;
+	}
+
+	@SuppressWarnings("unused")
+	static class Second {
+		@Embedded("thrd") Third third;
+		Third third2;
+	}
+
+	@SuppressWarnings("unused")
+	static class Third {
+		String value;
+	}
+
+}

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

@@ -29,7 +29,7 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
  * @author Jens Schauder
  */
 @UtilityClass
-class PropertyPathUtils {
+class PropertyPathTestingUtils {
 
 	static PersistentPropertyPath<RelationalPersistentProperty> toPath(String path, Class source,
 			RelationalMappingContext context) {

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

@@ -1,89 +0,0 @@
-/*
- * Copyright 2017-2019 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;
-
-import static org.assertj.core.api.Assertions.*;
-
-import org.junit.Test;
-
-/**
- * Unit tests for the {@link SelectBuilder}.
- *
- * @author Jens Schauder
- */
-public class SelectBuilderUnitTests {
-
-	@Test // DATAJDBC-112
-	public void simplestSelect() {
-
-		String sql = new SelectBuilder("mytable") //
-				.column(cb -> cb.tableAlias("mytable").column("mycolumn").as("myalias")) //
-				.build();
-
-		assertThat(sql).isEqualTo("SELECT mytable.mycolumn AS myalias FROM mytable");
-	}
-
-	@Test // DATAJDBC-112
-	public void columnWithoutTableAlias() {
-
-		String sql = new SelectBuilder("mytable") //
-				.column(cb -> cb.column("mycolumn").as("myalias")) //
-				.build();
-
-		assertThat(sql).isEqualTo("SELECT mycolumn AS myalias FROM mytable");
-	}
-
-	@Test // DATAJDBC-112
-	public void whereClause() {
-
-		String sql = new SelectBuilder("mytable") //
-				.column(cb -> cb.tableAlias("mytable").column("mycolumn").as("myalias")) //
-				.where(cb -> cb.tableAlias("mytable").column("mycolumn").eq().variable("var")).build();
-
-		assertThat(sql).isEqualTo("SELECT mytable.mycolumn AS myalias FROM mytable WHERE mytable.mycolumn = :var");
-	}
-
-	@Test // DATAJDBC-112
-	public void multipleColumnsSelect() {
-
-		String sql = new SelectBuilder("mytable") //
-				.column(cb -> cb.tableAlias("mytable").column("one").as("oneAlias")) //
-				.column(cb -> cb.tableAlias("mytable").column("two").as("twoAlias")) //
-				.build();
-
-		assertThat(sql).isEqualTo("SELECT mytable.one AS oneAlias, mytable.two AS twoAlias FROM mytable");
-	}
-
-	@Test // DATAJDBC-112
-	public void join() {
-		String sql = new SelectBuilder("mytable") //
-				.column(cb -> cb.tableAlias("mytable").column("mycolumn").as("myalias")) //
-				.join(jb -> jb.table("other").as("o").where("oid").eq().column("mytable", "id")).build();
-
-		assertThat(sql).isEqualTo("SELECT mytable.mycolumn AS myalias FROM mytable JOIN other AS o ON o.oid = mytable.id");
-	}
-
-	@Test // DATAJDBC-112
-	public void outerJoin() {
-		String sql = new SelectBuilder("mytable") //
-				.column(cb -> cb.tableAlias("mytable").column("mycolumn").as("myalias")) //
-				.join(jb -> jb.rightOuter().table("other").as("o").where("oid").eq().column("mytable", "id")).build();
-
-		assertThat(sql)
-				.isEqualTo("SELECT mytable.mycolumn AS myalias FROM mytable RIGHT OUTER JOIN other AS o ON o.oid = mytable.id");
-	}
-
-}

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

@@ -84,9 +84,10 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 
 			SqlGenerator sqlGenerator = configureSqlGenerator(contextualNamingStrategy);
 
-			String sql = sqlGenerator.createDeleteByPath(getPath("ref", DummyEntity.class));
+			String sql = sqlGenerator.createDeleteByPath(getPath("ref"));
 
-			assertThat(sql).isEqualTo("DELETE FROM " + user + ".referenced_entity WHERE " + "dummy_entity = :rootId");
+			assertThat(sql).isEqualTo(
+					"DELETE FROM " + user + ".referenced_entity WHERE " + user + ".referenced_entity.dummy_entity = :rootId");
 		});
 	}
 
@@ -97,13 +98,13 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 
 			SqlGenerator sqlGenerator = configureSqlGenerator(contextualNamingStrategy);
 
-			String sql = sqlGenerator.createDeleteByPath(getPath("ref.further", DummyEntity.class));
+			String sql = sqlGenerator.createDeleteByPath(getPath("ref.further"));
 
 			assertThat(sql).isEqualTo( //
 					"DELETE FROM " + user + ".second_level_referenced_entity " //
-							+ "WHERE " + "referenced_entity IN " //
-							+ "(SELECT l1id FROM " + user + ".referenced_entity " //
-							+ "WHERE " + "dummy_entity = :rootId)");
+							+ "WHERE " + user + ".second_level_referenced_entity.referenced_entity IN " //
+							+ "(SELECT " + user + ".referenced_entity.l1id FROM " + user + ".referenced_entity " //
+							+ "WHERE " + user + ".referenced_entity.dummy_entity = :rootId)");
 		});
 	}
 
@@ -127,10 +128,10 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 
 			SqlGenerator sqlGenerator = configureSqlGenerator(contextualNamingStrategy);
 
-			String sql = sqlGenerator.createDeleteAllSql(getPath("ref", DummyEntity.class));
+			String sql = sqlGenerator.createDeleteAllSql(getPath("ref"));
 
 			assertThat(sql).isEqualTo( //
-					"DELETE FROM " + user + ".referenced_entity WHERE " + "dummy_entity IS NOT NULL");
+					"DELETE FROM " + user + ".referenced_entity WHERE " + user + ".referenced_entity.dummy_entity IS NOT NULL");
 		});
 	}
 
@@ -141,18 +142,18 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 
 			SqlGenerator sqlGenerator = configureSqlGenerator(contextualNamingStrategy);
 
-			String sql = sqlGenerator.createDeleteAllSql(getPath("ref.further", DummyEntity.class));
+			String sql = sqlGenerator.createDeleteAllSql(getPath("ref.further"));
 
 			assertThat(sql).isEqualTo( //
 					"DELETE FROM " + user + ".second_level_referenced_entity " //
-							+ "WHERE " + "referenced_entity IN " //
-							+ "(SELECT l1id FROM " + user + ".referenced_entity " //
-							+ "WHERE " + "dummy_entity IS NOT NULL)");
+							+ "WHERE " + user + ".second_level_referenced_entity.referenced_entity IN " //
+							+ "(SELECT " + user + ".referenced_entity.l1id FROM " + user + ".referenced_entity " //
+							+ "WHERE " + user + ".referenced_entity.dummy_entity IS NOT NULL)");
 		});
 	}
 
-	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path, Class<DummyEntity> baseType) {
-		return PersistentPropertyPathTestUtils.getPath(this.context, path, baseType);
+	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path) {
+		return PersistentPropertyPathTestUtils.getPath(this.context, path, DummyEntity.class);
 	}
 
 	/**
@@ -214,6 +215,7 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 		return new SqlGenerator(context, persistentEntity, new SqlGeneratorSource(context));
 	}
 
+	@SuppressWarnings("unused")
 	static class DummyEntity {
 
 		@Id Long id;
@@ -221,6 +223,7 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 		ReferencedEntity ref;
 	}
 
+	@SuppressWarnings("unused")
 	static class ReferencedEntity {
 
 		@Id Long l1id;
@@ -228,6 +231,7 @@ public class SqlGeneratorContextBasedNamingStrategyUnitTests {
 		SecondLevelReferencedEntity further;
 	}
 
+	@SuppressWarnings("unused")
 	static class SecondLevelReferencedEntity {
 
 		@Id Long l2id;

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

@@ -16,9 +16,11 @@
 package org.springframework.data.jdbc.core;
 
 import static java.util.Collections.*;
+import static org.assertj.core.api.Assertions.*;
 
 import org.assertj.core.api.SoftAssertions;
 import org.junit.Before;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.core.mapping.JdbcMappingContext;
@@ -26,6 +28,7 @@ import org.springframework.data.relational.core.mapping.Column;
 import org.springframework.data.relational.core.mapping.Embedded;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
+import org.springframework.data.relational.core.sql.Aliased;
 
 /**
  * Unit tests for the {@link SqlGenerator} in a context of the {@link Embedded} annotation.
@@ -34,6 +37,7 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentEnti
  */
 public class SqlGeneratorEmbeddedUnitTests {
 
+	private RelationalMappingContext context = new JdbcMappingContext();
 	private SqlGenerator sqlGenerator;
 
 	@Before
@@ -42,7 +46,6 @@ public class SqlGeneratorEmbeddedUnitTests {
 	}
 
 	SqlGenerator createSqlGenerator(Class<?> type) {
-		RelationalMappingContext context = new JdbcMappingContext();
 		RelationalPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(type);
 		return new SqlGenerator(context, persistentEntity, new SqlGeneratorSource(context));
 	}
@@ -52,44 +55,31 @@ public class SqlGeneratorEmbeddedUnitTests {
 		final String sql = sqlGenerator.getFindOne();
 
 		SoftAssertions softAssertions = new SoftAssertions();
-		softAssertions.assertThat(sql)
-				.startsWith("SELECT")
-				.contains("dummy_entity.id1 AS id1")
-				.contains("dummy_entity.test AS test")
-				.contains("dummy_entity.attr1 AS attr1")
-				.contains("dummy_entity.attr2 AS attr2")
-				.contains("dummy_entity.prefix2_attr1 AS prefix2_attr1")
-				.contains("dummy_entity.prefix2_attr2 AS prefix2_attr2")
-				.contains("dummy_entity.prefix_test AS prefix_test")
-				.contains("dummy_entity.prefix_attr1 AS prefix_attr1")
-				.contains("dummy_entity.prefix_attr2 AS prefix_attr2")
+		softAssertions.assertThat(sql).startsWith("SELECT").contains("dummy_entity.id1 AS id1")
+				.contains("dummy_entity.test AS test").contains("dummy_entity.attr1 AS attr1")
+				.contains("dummy_entity.attr2 AS attr2").contains("dummy_entity.prefix2_attr1 AS prefix2_attr1")
+				.contains("dummy_entity.prefix2_attr2 AS prefix2_attr2").contains("dummy_entity.prefix_test AS prefix_test")
+				.contains("dummy_entity.prefix_attr1 AS prefix_attr1").contains("dummy_entity.prefix_attr2 AS prefix_attr2")
 				.contains("dummy_entity.prefix_prefix2_attr1 AS prefix_prefix2_attr1")
-				.contains("dummy_entity.prefix_prefix2_attr2 AS prefix_prefix2_attr2")
-				.contains("WHERE dummy_entity.id1 = :id")
+				.contains("dummy_entity.prefix_prefix2_attr2 AS prefix_prefix2_attr2").contains("WHERE dummy_entity.id1 = :id")
 				.doesNotContain("JOIN").doesNotContain("embeddable");
 		softAssertions.assertAll();
 	}
 
 	@Test // DATAJDBC-111
-		public void findAll() {
-			final String sql = sqlGenerator.getFindAll();
-
-			SoftAssertions softAssertions = new SoftAssertions();
-			softAssertions.assertThat(sql)
-					.startsWith("SELECT")
-					.contains("dummy_entity.id1 AS id1")
-					.contains("dummy_entity.test AS test")
-					.contains("dummy_entity.attr1 AS attr1")
-					.contains("dummy_entity.attr2 AS attr2")
-					.contains("dummy_entity.prefix2_attr1 AS prefix2_attr1")
-					.contains("dummy_entity.prefix2_attr2 AS prefix2_attr2")
-					.contains("dummy_entity.prefix_test AS prefix_test")
-					.contains("dummy_entity.prefix_attr1 AS prefix_attr1")
-					.contains("dummy_entity.prefix_attr2 AS prefix_attr2")
-					.contains("dummy_entity.prefix_prefix2_attr1 AS prefix_prefix2_attr1")
-					.contains("dummy_entity.prefix_prefix2_attr2 AS prefix_prefix2_attr2")
-					.doesNotContain("JOIN").doesNotContain("embeddable");
-			softAssertions.assertAll();
+	public void findAll() {
+		final String sql = sqlGenerator.getFindAll();
+
+		SoftAssertions softAssertions = new SoftAssertions();
+		softAssertions.assertThat(sql).startsWith("SELECT").contains("dummy_entity.id1 AS id1")
+				.contains("dummy_entity.test AS test").contains("dummy_entity.attr1 AS attr1")
+				.contains("dummy_entity.attr2 AS attr2").contains("dummy_entity.prefix2_attr1 AS prefix2_attr1")
+				.contains("dummy_entity.prefix2_attr2 AS prefix2_attr2").contains("dummy_entity.prefix_test AS prefix_test")
+				.contains("dummy_entity.prefix_attr1 AS prefix_attr1").contains("dummy_entity.prefix_attr2 AS prefix_attr2")
+				.contains("dummy_entity.prefix_prefix2_attr1 AS prefix_prefix2_attr1")
+				.contains("dummy_entity.prefix_prefix2_attr2 AS prefix_prefix2_attr2").doesNotContain("JOIN")
+				.doesNotContain("embeddable");
+		softAssertions.assertAll();
 	}
 
 	@Test // DATAJDBC-111
@@ -97,21 +87,14 @@ public class SqlGeneratorEmbeddedUnitTests {
 		final String sql = sqlGenerator.getFindAllInList();
 
 		SoftAssertions softAssertions = new SoftAssertions();
-		softAssertions.assertThat(sql)
-				.startsWith("SELECT")
-				.contains("dummy_entity.id1 AS id1")
-				.contains("dummy_entity.test AS test")
-				.contains("dummy_entity.attr1 AS attr1")
-				.contains("dummy_entity.attr2 AS attr2")
-				.contains("dummy_entity.prefix2_attr1 AS prefix2_attr1")
-				.contains("dummy_entity.prefix2_attr2 AS prefix2_attr2")
-				.contains("dummy_entity.prefix_test AS prefix_test")
-				.contains("dummy_entity.prefix_attr1 AS prefix_attr1")
-				.contains("dummy_entity.prefix_attr2 AS prefix_attr2")
+		softAssertions.assertThat(sql).startsWith("SELECT").contains("dummy_entity.id1 AS id1")
+				.contains("dummy_entity.test AS test").contains("dummy_entity.attr1 AS attr1")
+				.contains("dummy_entity.attr2 AS attr2").contains("dummy_entity.prefix2_attr1 AS prefix2_attr1")
+				.contains("dummy_entity.prefix2_attr2 AS prefix2_attr2").contains("dummy_entity.prefix_test AS prefix_test")
+				.contains("dummy_entity.prefix_attr1 AS prefix_attr1").contains("dummy_entity.prefix_attr2 AS prefix_attr2")
 				.contains("dummy_entity.prefix_prefix2_attr1 AS prefix_prefix2_attr1")
 				.contains("dummy_entity.prefix_prefix2_attr2 AS prefix_prefix2_attr2")
-				.contains("WHERE dummy_entity.id1 in(:ids)")
-				.doesNotContain("JOIN").doesNotContain("embeddable");
+				.contains("WHERE dummy_entity.id1 IN (:ids)").doesNotContain("JOIN").doesNotContain("embeddable");
 		softAssertions.assertAll();
 	}
 
@@ -120,18 +103,9 @@ public class SqlGeneratorEmbeddedUnitTests {
 		final String sql = sqlGenerator.getInsert(emptySet());
 
 		SoftAssertions softAssertions = new SoftAssertions();
-		softAssertions.assertThat(sql)
-				.startsWith("INSERT INTO")
-				.contains("dummy_entity")
-				.contains(":test")
-				.contains(":attr1")
-				.contains(":attr2")
-				.contains(":prefix2_attr1")
-				.contains(":prefix2_attr2")
-				.contains(":prefix_test")
-				.contains(":prefix_attr1")
-				.contains(":prefix_attr2")
-				.contains(":prefix_prefix2_attr1")
+		softAssertions.assertThat(sql).startsWith("INSERT INTO").contains("dummy_entity").contains(":test")
+				.contains(":attr1").contains(":attr2").contains(":prefix2_attr1").contains(":prefix2_attr2")
+				.contains(":prefix_test").contains(":prefix_attr1").contains(":prefix_attr2").contains(":prefix_prefix2_attr1")
 				.contains(":prefix_prefix2_attr2");
 		softAssertions.assertAll();
 	}
@@ -141,48 +115,166 @@ public class SqlGeneratorEmbeddedUnitTests {
 		final String sql = sqlGenerator.getUpdate();
 
 		SoftAssertions softAssertions = new SoftAssertions();
-		softAssertions.assertThat(sql)
-				.startsWith("UPDATE")
-				.contains("dummy_entity")
-				.contains("test = :test")
-				.contains("attr1 = :attr1")
-				.contains("attr2 = :attr2")
-				.contains("prefix2_attr1 = :prefix2_attr1")
-				.contains("prefix2_attr2 = :prefix2_attr2")
-				.contains("prefix_test = :prefix_test")
-				.contains("prefix_attr1 = :prefix_attr1")
-				.contains("prefix_attr2 = :prefix_attr2")
+		softAssertions.assertThat(sql).startsWith("UPDATE").contains("dummy_entity").contains("test = :test")
+				.contains("attr1 = :attr1").contains("attr2 = :attr2").contains("prefix2_attr1 = :prefix2_attr1")
+				.contains("prefix2_attr2 = :prefix2_attr2").contains("prefix_test = :prefix_test")
+				.contains("prefix_attr1 = :prefix_attr1").contains("prefix_attr2 = :prefix_attr2")
 				.contains("prefix_prefix2_attr1 = :prefix_prefix2_attr1")
 				.contains("prefix_prefix2_attr2 = :prefix_prefix2_attr2");
 		softAssertions.assertAll();
 	}
 
+	@Test // DATAJDBC-340
+	@Ignore // this is just broken right now
+	public void deleteByPath() {
+
+		final String sql = sqlGenerator
+				.createDeleteByPath(PropertyPathTestingUtils.toPath("embedded.other", DummyEntity2.class, context));
+
+		assertThat(sql).containsSequence("DELETE FROM other_entity", //
+				"WHERE", //
+				"embedded_with_reference IN (", //
+				"SELECT ", //
+				"id ", //
+				"FROM", //
+				"dummy_entity2", //
+				"WHERE", //
+				"embedded_with_reference = :rootId");
+	}
+
+	@Test // DATAJDBC-340
+	public void noJoinForEmbedded() {
+
+		SqlGenerator.Join join = generateJoin("embeddable", DummyEntity.class);
+
+		assertThat(join).isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void columnForEmbeddedProperty() {
+
+		assertThat(generatedColumn("embeddable.test", DummyEntity.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias)
+				.containsExactly("test", "dummy_entity", null, "test");
+	}
+
+	@Test // DATAJDBC-340
+	public void noColumnForEmbedded() {
+
+		assertThat(generatedColumn("embeddable", DummyEntity.class)) //
+				.isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void noJoinForPrefixedEmbedded() {
+
+		SqlGenerator.Join join = generateJoin("prefixedEmbeddable", DummyEntity.class);
+
+		assertThat(join).isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void columnForPrefixedEmbeddedProperty() {
+
+		assertThat(generatedColumn("prefixedEmbeddable.test", DummyEntity.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias)
+				.containsExactly("prefix_test", "dummy_entity", null, "prefix_test");
+	}
+
+	@Test // DATAJDBC-340
+	public void noJoinForCascadedEmbedded() {
+
+		SqlGenerator.Join join = generateJoin("embeddable.embeddable", DummyEntity.class);
+
+		assertThat(join).isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void columnForCascadedEmbeddedProperty() {
+
+		assertThat(generatedColumn("embeddable.embeddable.attr1", DummyEntity.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias)
+				.containsExactly("attr1", "dummy_entity", null, "attr1");
+	}
+
+	@Test // DATAJDBC-340
+	public void joinForEmbeddedWithReference() {
+
+		SqlGenerator.Join join = generateJoin("embedded.other", DummyEntity2.class);
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			softly.assertThat(join.getJoinTable().getName()).isEqualTo("other_entity");
+			softly.assertThat(join.getJoinColumn().getTable()).isEqualTo(join.getJoinTable());
+			softly.assertThat(join.getJoinColumn().getName()).isEqualTo("embedded_with_reference");
+			softly.assertThat(join.getParentId().getName()).isEqualTo("id");
+			softly.assertThat(join.getParentId().getTable().getName()).isEqualTo("dummy_entity2");
+		});
+	}
+
+	@Test // DATAJDBC-340
+	public void columnForEmbeddedWithReferenceProperty() {
+
+		assertThat(generatedColumn("embedded.other.value", DummyEntity2.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias)
+				.containsExactly("value", "other_entity", "prefix_other", "prefix_other_value");
+	}
+
+	private SqlGenerator.Join generateJoin(String path, Class<?> type) {
+		return createSqlGenerator(type)
+				.getJoin(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+	}
+
+	private String getAlias(Object maybeAliased) {
+
+		if (maybeAliased instanceof Aliased) {
+			return ((Aliased) maybeAliased).getAlias();
+		}
+		return null;
+	}
+
+	private org.springframework.data.relational.core.sql.Column generatedColumn(String path, Class<?> type) {
+		return createSqlGenerator(type)
+				.getColumn(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+	}
+
 	@SuppressWarnings("unused")
 	static class DummyEntity {
 
-		@Column("id1")
-		@Id
-		Long id;
+		@Column("id1") @Id Long id;
 
-		@Embedded("prefix_")
-		CascadedEmbedded prefixedEmbeddable;
+		@Embedded("prefix_") CascadedEmbedded prefixedEmbeddable;
 
-		@Embedded
-		CascadedEmbedded embeddable;
+		@Embedded CascadedEmbedded embeddable;
 	}
 
 	@SuppressWarnings("unused")
-	static class CascadedEmbedded
-	{
+	static class CascadedEmbedded {
 		String test;
 		@Embedded("prefix2_") Embeddable prefixedEmbeddable;
 		@Embedded Embeddable embeddable;
 	}
 
 	@SuppressWarnings("unused")
-	static class Embeddable
-	{
+	static class Embeddable {
 		Long attr1;
 		String attr2;
 	}
+
+	@SuppressWarnings("unused")
+	static class DummyEntity2 {
+
+		@Id Long id;
+
+		@Embedded("prefix_") EmbeddedWithReference embedded;
+	}
+
+	static class EmbeddedWithReference {
+		OtherEntity other;
+	}
+
+	static class OtherEntity {
+		String value;
+	}
+
 }

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

@@ -112,10 +112,10 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 
 		SqlGenerator sqlGenerator = configureSqlGenerator(fixedCustomTablePrefixStrategy);
 
-		String sql = sqlGenerator.createDeleteByPath(getPath("ref", DummyEntity.class));
+		String sql = sqlGenerator.createDeleteByPath(getPath("ref"));
 
-		assertThat(sql).isEqualTo(
-				"DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity " + "WHERE dummy_entity = :rootId");
+		assertThat(sql).isEqualTo("DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity "
+				+ "WHERE FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity.dummy_entity = :rootId");
 	}
 
 	@Test // DATAJDBC-107
@@ -123,11 +123,13 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 
 		SqlGenerator sqlGenerator = configureSqlGenerator(fixedCustomTablePrefixStrategy);
 
-		String sql = sqlGenerator.createDeleteByPath(getPath("ref.further", DummyEntity.class));
+		String sql = sqlGenerator.createDeleteByPath(getPath("ref.further"));
 
 		assertThat(sql).isEqualTo("DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_SecondLevelReferencedEntity "
-				+ "WHERE referenced_entity IN " + "(SELECT FixedCustomPropertyPrefix_l1id "
-				+ "FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity " + "WHERE dummy_entity = :rootId)");
+				+ "WHERE FixedCustomSchema.FixedCustomTablePrefix_SecondLevelReferencedEntity.referenced_entity IN "
+				+ "(SELECT FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity.FixedCustomPropertyPrefix_l1id "
+				+ "FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity "
+				+ "WHERE FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity.dummy_entity = :rootId)");
 	}
 
 	@Test // DATAJDBC-107
@@ -145,10 +147,10 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 
 		SqlGenerator sqlGenerator = configureSqlGenerator(fixedCustomTablePrefixStrategy);
 
-		String sql = sqlGenerator.createDeleteAllSql(getPath("ref", DummyEntity.class));
+		String sql = sqlGenerator.createDeleteAllSql(getPath("ref"));
 
-		assertThat(sql).isEqualTo(
-				"DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity " + "WHERE dummy_entity IS NOT NULL");
+		assertThat(sql).isEqualTo("DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity "
+				+ "WHERE FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity.dummy_entity IS NOT NULL");
 	}
 
 	@Test // DATAJDBC-107
@@ -156,11 +158,13 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 
 		SqlGenerator sqlGenerator = configureSqlGenerator(fixedCustomTablePrefixStrategy);
 
-		String sql = sqlGenerator.createDeleteAllSql(getPath("ref.further", DummyEntity.class));
+		String sql = sqlGenerator.createDeleteAllSql(getPath("ref.further"));
 
 		assertThat(sql).isEqualTo("DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_SecondLevelReferencedEntity "
-				+ "WHERE referenced_entity IN " + "(SELECT FixedCustomPropertyPrefix_l1id "
-				+ "FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity " + "WHERE dummy_entity IS NOT NULL)");
+				+ "WHERE FixedCustomSchema.FixedCustomTablePrefix_SecondLevelReferencedEntity.referenced_entity IN "
+				+ "(SELECT FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity.FixedCustomPropertyPrefix_l1id "
+				+ "FROM FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity "
+				+ "WHERE FixedCustomSchema.FixedCustomTablePrefix_ReferencedEntity.dummy_entity IS NOT NULL)");
 	}
 
 	@Test // DATAJDBC-113
@@ -171,17 +175,15 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 		String sql = sqlGenerator.getDeleteByList();
 
 		assertThat(sql).isEqualTo(
-				"DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_DummyEntity WHERE FixedCustomPropertyPrefix_id IN (:ids)");
+				"DELETE FROM FixedCustomSchema.FixedCustomTablePrefix_DummyEntity WHERE FixedCustomSchema.FixedCustomTablePrefix_DummyEntity.FixedCustomPropertyPrefix_id IN (:ids)");
 	}
 
-	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path, Class<?> baseType) {
-		return PersistentPropertyPathTestUtils.getPath(context, path, baseType);
+	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path) {
+		return PersistentPropertyPathTestUtils.getPath(context, path, DummyEntity.class);
 	}
 
 	/**
 	 * Plug in a custom {@link NamingStrategy} for this test case.
-	 *
-	 * @param namingStrategy
 	 */
 	private SqlGenerator configureSqlGenerator(NamingStrategy namingStrategy) {
 
@@ -190,6 +192,7 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 		return new SqlGenerator(context, persistentEntity, new SqlGeneratorSource(context));
 	}
 
+	@SuppressWarnings("unused")
 	static class DummyEntity {
 
 		@Id Long id;
@@ -197,6 +200,7 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 		ReferencedEntity ref;
 	}
 
+	@SuppressWarnings("unused")
 	static class ReferencedEntity {
 
 		@Id Long l1id;
@@ -204,6 +208,7 @@ public class SqlGeneratorFixedNamingStrategyUnitTests {
 		SecondLevelReferencedEntity further;
 	}
 
+	@SuppressWarnings("unused")
 	static class SecondLevelReferencedEntity {
 
 		@Id Long l2id;

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

@@ -35,6 +35,8 @@ import org.springframework.data.relational.core.mapping.NamingStrategy;
 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.relational.core.sql.Aliased;
+import org.springframework.data.relational.core.sql.Table;
 
 /**
  * Unit tests for the {@link SqlGenerator}.
@@ -46,19 +48,17 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
  */
 public class SqlGeneratorUnitTests {
 
-	private SqlGenerator sqlGenerator;
-	private RelationalMappingContext context = new JdbcMappingContext();
+	SqlGenerator sqlGenerator;
+	NamingStrategy namingStrategy = new PrefixingNamingStrategy();
+	RelationalMappingContext context = new JdbcMappingContext(namingStrategy);
 
 	@Before
 	public void setUp() {
-
 		this.sqlGenerator = createSqlGenerator(DummyEntity.class);
 	}
 
 	SqlGenerator createSqlGenerator(Class<?> type) {
 
-		NamingStrategy namingStrategy = new PrefixingNamingStrategy();
-		RelationalMappingContext context = new JdbcMappingContext(namingStrategy);
 		RelationalPersistentEntity<?> persistentEntity = context.getRequiredPersistentEntity(type);
 
 		return new SqlGenerator(context, persistentEntity, new SqlGeneratorSource(context));
@@ -87,18 +87,18 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-112
 	public void cascadingDeleteFirstLevel() {
 
-		String sql = sqlGenerator.createDeleteByPath(getPath("ref"));
+		String sql = sqlGenerator.createDeleteByPath(getPath("ref", DummyEntity.class));
 
-		assertThat(sql).isEqualTo("DELETE FROM referenced_entity WHERE dummy_entity = :rootId");
+		assertThat(sql).isEqualTo("DELETE FROM referenced_entity WHERE referenced_entity.dummy_entity = :rootId");
 	}
 
 	@Test // DATAJDBC-112
-	public void cascadingDeleteAllSecondLevel() {
+	public void cascadingDeleteByPathSecondLevel() {
 
-		String sql = sqlGenerator.createDeleteByPath(getPath("ref.further"));
+		String sql = sqlGenerator.createDeleteByPath(getPath("ref.further", DummyEntity.class));
 
 		assertThat(sql).isEqualTo(
-				"DELETE FROM second_level_referenced_entity WHERE referenced_entity IN (SELECT x_l1id FROM referenced_entity WHERE dummy_entity = :rootId)");
+				"DELETE FROM second_level_referenced_entity WHERE second_level_referenced_entity.referenced_entity IN (SELECT referenced_entity.x_l1id FROM referenced_entity WHERE referenced_entity.dummy_entity = :rootId)");
 	}
 
 	@Test // DATAJDBC-112
@@ -112,34 +112,34 @@ public class SqlGeneratorUnitTests {
 	@Test // DATAJDBC-112
 	public void cascadingDeleteAllFirstLevel() {
 
-		String sql = sqlGenerator.createDeleteAllSql(getPath("ref"));
+		String sql = sqlGenerator.createDeleteAllSql(getPath("ref", DummyEntity.class));
 
-		assertThat(sql).isEqualTo("DELETE FROM referenced_entity WHERE dummy_entity IS NOT NULL");
+		assertThat(sql).isEqualTo("DELETE FROM referenced_entity WHERE referenced_entity.dummy_entity IS NOT NULL");
 	}
 
 	@Test // DATAJDBC-112
-	public void cascadingDeleteSecondLevel() {
+	public void cascadingDeleteAllSecondLevel() {
 
-		String sql = sqlGenerator.createDeleteAllSql(getPath("ref.further"));
+		String sql = sqlGenerator.createDeleteAllSql(getPath("ref.further", DummyEntity.class));
 
 		assertThat(sql).isEqualTo(
-				"DELETE FROM second_level_referenced_entity WHERE referenced_entity IN (SELECT x_l1id FROM referenced_entity WHERE dummy_entity IS NOT NULL)");
+				"DELETE FROM second_level_referenced_entity WHERE second_level_referenced_entity.referenced_entity IN (SELECT referenced_entity.x_l1id FROM referenced_entity WHERE referenced_entity.dummy_entity IS NOT NULL)");
 	}
 
 	@Test // DATAJDBC-227
 	public void deleteAllMap() {
 
-		String sql = sqlGenerator.createDeleteAllSql(getPath("mappedElements"));
+		String sql = sqlGenerator.createDeleteAllSql(getPath("mappedElements", DummyEntity.class));
 
-		assertThat(sql).isEqualTo("DELETE FROM element WHERE dummy_entity IS NOT NULL");
+		assertThat(sql).isEqualTo("DELETE FROM element WHERE element.dummy_entity IS NOT NULL");
 	}
 
 	@Test // DATAJDBC-227
 	public void deleteMapByPath() {
 
-		String sql = sqlGenerator.createDeleteByPath(getPath("mappedElements"));
+		String sql = sqlGenerator.createDeleteByPath(getPath("mappedElements", DummyEntity.class));
 
-		assertThat(sql).isEqualTo("DELETE FROM element WHERE dummy_entity = :rootId");
+		assertThat(sql).isEqualTo("DELETE FROM element WHERE element.dummy_entity = :rootId");
 	}
 
 	@Test // DATAJDBC-131, DATAJDBC-111
@@ -148,14 +148,18 @@ public class SqlGeneratorUnitTests {
 		// this would get called when ListParent is the element type of a Set
 		String sql = sqlGenerator.getFindAllByProperty("back-ref", null, false);
 
-		assertThat(sql).isEqualTo("SELECT dummy_entity.id1 AS id1, dummy_entity.x_name AS x_name, " //
-				+ "dummy_entity.x_other AS x_other, " //
-				+ "ref.x_l1id AS ref_x_l1id, ref.x_content AS ref_x_content, "
-				+ "ref_further.x_l2id AS ref_further_x_l2id, ref_further.x_something AS ref_further_x_something " //
-				+ "FROM dummy_entity " //
-				+ "LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1  " //
-				+ "LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = referenced_entity.x_l1id " //
-				+ "WHERE back-ref = :back-ref");
+		assertThat(sql).contains("SELECT", //
+				"dummy_entity.id1 AS id1", //
+				"dummy_entity.x_name AS x_name", //
+				"dummy_entity.x_other AS x_other", //
+				"ref.x_l1id AS ref_x_l1id", //
+				"ref.x_content AS ref_x_content", //
+				"ref_further.x_l2id AS ref_further_x_l2id", //
+				"ref_further.x_something AS ref_further_x_something", //
+				"FROM dummy_entity ", //
+				"LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1", //
+				"LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id", //
+				"WHERE dummy_entity.back-ref = :back-ref");
 	}
 
 	@Test // DATAJDBC-131, DATAJDBC-111
@@ -170,9 +174,9 @@ public class SqlGeneratorUnitTests {
 				+ "ref_further.x_l2id AS ref_further_x_l2id, ref_further.x_something AS ref_further_x_something, " //
 				+ "dummy_entity.key-column AS key-column " //
 				+ "FROM dummy_entity " //
-				+ "LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1  " //
-				+ "LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = referenced_entity.x_l1id " //
-				+ "WHERE back-ref = :back-ref");
+				+ "LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1 " //
+				+ "LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id " //
+				+ "WHERE dummy_entity.back-ref = :back-ref");
 	}
 
 	@Test(expected = IllegalArgumentException.class) // DATAJDBC-130
@@ -192,9 +196,9 @@ public class SqlGeneratorUnitTests {
 				+ "ref_further.x_l2id AS ref_further_x_l2id, ref_further.x_something AS ref_further_x_something, " //
 				+ "dummy_entity.key-column AS key-column " //
 				+ "FROM dummy_entity " //
-				+ "LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1  " //
-				+ "LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = referenced_entity.x_l1id " //
-				+ "WHERE back-ref = :back-ref " + "ORDER BY key-column");
+				+ "LEFT OUTER JOIN referenced_entity AS ref ON ref.dummy_entity = dummy_entity.id1 " //
+				+ "LEFT OUTER JOIN second_level_referenced_entity AS ref_further ON ref_further.referenced_entity = ref.x_l1id " //
+				+ "WHERE dummy_entity.back-ref = :back-ref " + "ORDER BY key-column");
 	}
 
 	@Test // DATAJDBC-264
@@ -214,9 +218,8 @@ public class SqlGeneratorUnitTests {
 
 		String insert = sqlGenerator.getInsert(emptySet());
 
-		assertThat(insert).isEqualTo("INSERT INTO entity_with_quoted_column_name " +
-				"(\"test_@123\") " +
-				"VALUES (:test_123)");
+		assertThat(insert)
+				.isEqualTo("INSERT INTO entity_with_quoted_column_name " + "(\"test_@123\") " + "VALUES (:test_123)");
 	}
 
 	@Test // DATAJDBC-266
@@ -249,7 +252,7 @@ public class SqlGeneratorUnitTests {
 		assertThat(sqlGenerator.getUpdate()).isEqualToIgnoringCase( //
 				"UPDATE entity_with_read_only_property " //
 						+ "SET x_name = :x_name " //
-						+ "WHERE x_id = :x_id" //
+						+ "WHERE entity_with_read_only_property.x_id = :x_id" //
 		);
 	}
 
@@ -260,9 +263,8 @@ public class SqlGeneratorUnitTests {
 
 		String update = sqlGenerator.getUpdate();
 
-		assertThat(update).isEqualTo("UPDATE entity_with_quoted_column_name " +
-				"SET \"test_@123\" = :test_123 " +
-				"WHERE \"test_@id\" = :test_id");
+		assertThat(update).isEqualTo("UPDATE entity_with_quoted_column_name " + "SET \"test_@123\" = :test_123 "
+				+ "WHERE entity_with_quoted_column_name.\"test_@id\" = :test_id");
 	}
 
 	@Test // DATAJDBC-324
@@ -294,13 +296,13 @@ public class SqlGeneratorUnitTests {
 
 		assertThat(sqlGenerator.getFindAllByProperty("back-ref", "key-column", true)).isEqualToIgnoringCase( //
 				"SELECT " //
-				+ "entity_with_read_only_property.x_id AS x_id, " //
-				+ "entity_with_read_only_property.x_name AS x_name, " //
-				+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value, " //
-				+ "entity_with_read_only_property.key-column AS key-column " //
-				+ "FROM entity_with_read_only_property " //
-				+ "WHERE back-ref = :back-ref " //
-				+ "ORDER BY key-column" //
+						+ "entity_with_read_only_property.x_id AS x_id, " //
+						+ "entity_with_read_only_property.x_name AS x_name, " //
+						+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value, " //
+						+ "entity_with_read_only_property.key-column AS key-column " //
+						+ "FROM entity_with_read_only_property " //
+						+ "WHERE entity_with_read_only_property.back-ref = :back-ref " //
+						+ "ORDER BY key-column" //
 		);
 	}
 
@@ -311,11 +313,11 @@ public class SqlGeneratorUnitTests {
 
 		assertThat(sqlGenerator.getFindAllInList()).isEqualToIgnoringCase( //
 				"SELECT " //
-				+ "entity_with_read_only_property.x_id AS x_id, " //
-				+ "entity_with_read_only_property.x_name AS x_name, " //
-				+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value " //
-				+ "FROM entity_with_read_only_property " //
-				+ "WHERE entity_with_read_only_property.x_id in(:ids)" //
+						+ "entity_with_read_only_property.x_id AS x_id, " //
+						+ "entity_with_read_only_property.x_name AS x_name, " //
+						+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value " //
+						+ "FROM entity_with_read_only_property " //
+						+ "WHERE entity_with_read_only_property.x_id IN (:ids)" //
 		);
 	}
 
@@ -326,23 +328,158 @@ public class SqlGeneratorUnitTests {
 
 		assertThat(sqlGenerator.getFindOne()).isEqualToIgnoringCase( //
 				"SELECT " //
-				+ "entity_with_read_only_property.x_id AS x_id, " //
-				+ "entity_with_read_only_property.x_name AS x_name, " //
-				+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value " //
-				+ "FROM entity_with_read_only_property " //
-				+ "WHERE entity_with_read_only_property.x_id = :id" //
+						+ "entity_with_read_only_property.x_id AS x_id, " //
+						+ "entity_with_read_only_property.x_name AS x_name, " //
+						+ "entity_with_read_only_property.x_read_only_value AS x_read_only_value " //
+						+ "FROM entity_with_read_only_property " //
+						+ "WHERE entity_with_read_only_property.x_id = :id" //
 		);
 	}
 
-	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path) {
-		return PersistentPropertyPathTestUtils.getPath(context, path, DummyEntity.class);
+	@Test // DATAJDBC-340
+	public void deletingLongChain() {
+
+		assertThat(
+				createSqlGenerator(Chain4.class).createDeleteByPath(getPath("chain3.chain2.chain1.chain0", Chain4.class))) //
+						.isEqualTo("DELETE FROM chain0 " + //
+								"WHERE chain0.chain1 IN (" + //
+								"SELECT chain1.x_one " + //
+								"FROM chain1 " + //
+								"WHERE chain1.chain2 IN (" + //
+								"SELECT chain2.x_two " + //
+								"FROM chain2 " + //
+								"WHERE chain2.chain3 IN (" + //
+								"SELECT chain3.x_three " + //
+								"FROM chain3 " + //
+								"WHERE chain3.chain4 = :rootId" + //
+								")))");
+	}
+
+	@Test // DATAJDBC-340
+	public void noJoinForSimpleColumn() {
+		assertThat(generateJoin("id", DummyEntity.class)).isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void joinForSimpleReference() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			SqlGenerator.Join join = generateJoin("ref", DummyEntity.class);
+			softly.assertThat(join.getJoinTable().getName()).isEqualTo("referenced_entity");
+			softly.assertThat(join.getJoinColumn().getTable()).isEqualTo(join.getJoinTable());
+			softly.assertThat(join.getJoinColumn().getName()).isEqualTo("dummy_entity");
+			softly.assertThat(join.getParentId().getName()).isEqualTo("id1");
+			softly.assertThat(join.getParentId().getTable().getName()).isEqualTo("dummy_entity");
+		});
+	}
+
+	@Test // DATAJDBC-340
+	public void noJoinForCollectionReference() {
+
+		SqlGenerator.Join join = generateJoin("elements", DummyEntity.class);
+
+		assertThat(join).isNull();
+
+	}
+
+	@Test // DATAJDBC-340
+	public void noJoinForMappedReference() {
+
+		SqlGenerator.Join join = generateJoin("mappedElements", DummyEntity.class);
+
+		assertThat(join).isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void joinForSecondLevelReference() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			SqlGenerator.Join join = generateJoin("ref.further", DummyEntity.class);
+			softly.assertThat(join.getJoinTable().getName()).isEqualTo("second_level_referenced_entity");
+			softly.assertThat(join.getJoinColumn().getTable()).isEqualTo(join.getJoinTable());
+			softly.assertThat(join.getJoinColumn().getName()).isEqualTo("referenced_entity");
+			softly.assertThat(join.getParentId().getName()).isEqualTo("x_l1id");
+			softly.assertThat(join.getParentId().getTable().getName()).isEqualTo("referenced_entity");
+		});
+	}
+
+	@Test // DATAJDBC-340
+	public void joinForOneToOneWithoutId() {
+
+		SoftAssertions.assertSoftly(softly -> {
+
+			SqlGenerator.Join join = generateJoin("child", ParentOfNoIdChild.class);
+			Table joinTable = join.getJoinTable();
+			softly.assertThat(joinTable.getName()).isEqualTo("no_id_child");
+			softly.assertThat(joinTable).isInstanceOf(Aliased.class);
+			softly.assertThat(((Aliased) joinTable).getAlias()).isEqualTo("child");
+			softly.assertThat(join.getJoinColumn().getTable()).isEqualTo(joinTable);
+			softly.assertThat(join.getJoinColumn().getName()).isEqualTo("parent_of_no_id_child");
+			softly.assertThat(join.getParentId().getName()).isEqualTo("x_id");
+			softly.assertThat(join.getParentId().getTable().getName()).isEqualTo("parent_of_no_id_child");
+
+		});
+	}
+
+	private SqlGenerator.Join generateJoin(String path, Class<?> type) {
+		return createSqlGenerator(type)
+				.getJoin(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+	}
+
+	@Test // DATAJDBC-340
+	public void simpleColumn() {
+
+		assertThat(generatedColumn("id", DummyEntity.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias)
+				.containsExactly("id1", "dummy_entity", null, "id1");
+	}
+
+	@Test // DATAJDBC-340
+	public void columnForIndirectProperty() {
+
+		assertThat(generatedColumn("ref.l1id", DummyEntity.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias) //
+				.containsExactly("x_l1id", "referenced_entity", "ref", "ref_x_l1id");
+	}
+
+	@Test // DATAJDBC-340
+	public void noColumnForReferencedEntity() {
+
+		assertThat(generatedColumn("ref", DummyEntity.class)).isNull();
+	}
+
+	@Test // DATAJDBC-340
+	public void columnForReferencedEntityWithoutId() {
+
+		assertThat(generatedColumn("child", ParentOfNoIdChild.class)) //
+				.extracting(c -> c.getName(), c -> c.getTable().getName(), c -> getAlias(c.getTable()), this::getAlias) //
+				.containsExactly("parent_of_no_id_child", "no_id_child", "child", "child_parent_of_no_id_child");
+	}
+
+	private String getAlias(Object maybeAliased) {
+
+		if (maybeAliased instanceof Aliased) {
+			return ((Aliased) maybeAliased).getAlias();
+		}
+		return null;
+	}
+
+	private org.springframework.data.relational.core.sql.Column generatedColumn(String path, Class<?> type) {
+
+		return createSqlGenerator(type)
+				.getColumn(new PersistentPropertyPathExtension(context, PropertyPathTestingUtils.toPath(path, type, context)));
+	}
+
+	private PersistentPropertyPath<RelationalPersistentProperty> getPath(String path, Class<?> baseType) {
+		return PersistentPropertyPathTestUtils.getPath(context, path, baseType);
 	}
 
 	@SuppressWarnings("unused")
 	static class DummyEntity {
 
-		@Column("id1")
-		@Id Long id;
+		@Column("id1") @Id Long id;
 		String name;
 		ReferencedEntity ref;
 		Set<Element> elements;
@@ -412,4 +549,38 @@ public class SqlGeneratorUnitTests {
 		@Id @Column("\"test_@id\"") Long id;
 		@Column("\"test_@123\"") String name;
 	}
+
+	@SuppressWarnings("unused")
+	static class Chain0 {
+		@Id Long zero;
+		String zeroValue;
+	}
+
+	@SuppressWarnings("unused")
+	static class Chain1 {
+		@Id Long one;
+		String oneValue;
+		Chain0 chain0;
+	}
+
+	@SuppressWarnings("unused")
+	static class Chain2 {
+		@Id Long two;
+		String twoValue;
+		Chain1 chain1;
+	}
+
+	@SuppressWarnings("unused")
+	static class Chain3 {
+		@Id Long three;
+		String threeValue;
+		Chain2 chain2;
+	}
+
+	@SuppressWarnings("unused")
+	static class Chain4 {
+		@Id Long four;
+		String fourValue;
+		Chain3 chain3;
+	}
 }

spring-data-jdbc/src/test/resources/logback.xml

@@ -7,8 +7,8 @@
 		</encoder>
 	</appender>
 
-	<logger name="org.springframework.data" level="info" />
-	<logger name="org.springframework.jdbc.core" level="trace" />
+	<!--<logger name="org.springframework.data" level="info" />-->
+	<!--<logger name="org.springframework.jdbc.core" level="trace" />-->
 	<!--<logger name="org.springframework.data.jdbc.mybatis.DummyEntityMapper" level="trace" />-->
 
 	<root level="warn">

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.core/JdbcAggregateTemplateIntegrationTests-hsql.sql

@@ -1,20 +1,93 @@
-CREATE TABLE LEGO_SET ( id1 BIGINT GENERATED BY DEFAULT AS IDENTITY(START WITH 1) PRIMARY KEY, NAME VARCHAR(30));
-CREATE TABLE MANUAL ( id2 BIGINT GENERATED BY DEFAULT AS IDENTITY(START WITH 1) PRIMARY KEY, LEGO_SET BIGINT, ALTERNATIVE BIGINT, CONTENT VARCHAR(2000));
+CREATE TABLE LEGO_SET
+(
+  id1  BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY,
+  NAME VARCHAR(30)
+);
+CREATE TABLE MANUAL
+(
+  id2         BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY,
+  LEGO_SET    BIGINT,
+  ALTERNATIVE BIGINT,
+  CONTENT     VARCHAR(2000)
+);
 
-ALTER TABLE MANUAL ADD FOREIGN KEY (LEGO_SET)
-REFERENCES LEGO_SET(id1);
+ALTER TABLE MANUAL
+  ADD FOREIGN KEY (LEGO_SET)
+    REFERENCES LEGO_SET (id1);
 
-CREATE TABLE ONE_TO_ONE_PARENT ( id3 BIGINT GENERATED BY DEFAULT AS IDENTITY(START WITH 1) PRIMARY KEY, content VARCHAR(30));
-CREATE TABLE Child_No_Id (ONE_TO_ONE_PARENT INTEGER PRIMARY KEY, content VARCHAR(30));
+CREATE TABLE ONE_TO_ONE_PARENT
+(
+  id3     BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY,
+  content VARCHAR(30)
+);
+CREATE TABLE Child_No_Id
+(
+  ONE_TO_ONE_PARENT INTEGER PRIMARY KEY,
+  content           VARCHAR(30)
+);
 
-CREATE TABLE LIST_PARENT ( id4 BIGINT GENERATED BY DEFAULT AS IDENTITY ( START WITH 1 ) PRIMARY KEY, NAME VARCHAR(100));
-CREATE TABLE ELEMENT_NO_ID ( content VARCHAR(100), LIST_PARENT_KEY BIGINT, LIST_PARENT BIGINT);
+CREATE TABLE LIST_PARENT
+(
+  id4  BIGINT GENERATED BY DEFAULT AS IDENTITY ( START WITH 1 ) PRIMARY KEY,
+  NAME VARCHAR(100)
+);
+CREATE TABLE ELEMENT_NO_ID
+(
+  content         VARCHAR(100),
+  LIST_PARENT_KEY BIGINT,
+  LIST_PARENT     BIGINT
+);
 ALTER TABLE ELEMENT_NO_ID
   ADD FOREIGN KEY (LIST_PARENT)
-    REFERENCES LIST_PARENT(id4);
+    REFERENCES LIST_PARENT (id4);
 
+CREATE TABLE ARRAY_OWNER
+(
+  ID               BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY,
+  DIGITS           VARCHAR(20) ARRAY[10] NOT NULL,
+  MULTIDIMENSIONAL VARCHAR(20) ARRAY[10] NULL
+);
 
-CREATE TABLE ARRAY_OWNER (ID BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY, DIGITS VARCHAR(20) ARRAY[10] NOT NULL, MULTIDIMENSIONAL VARCHAR(20) ARRAY[10] NULL);
+CREATE TABLE BYTE_ARRAY_OWNER
+(
+  ID          BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY,
+  BINARY_DATA VARBINARY(20) NOT NULL
+);
 
-CREATE TABLE BYTE_ARRAY_OWNER (ID BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 1) PRIMARY KEY, BINARY_DATA VARBINARY(20) NOT NULL);
+CREATE TABLE CHAIN4
+(
+  FOUR       BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 40) PRIMARY KEY,
+  FOUR_VALUE VARCHAR(20)
+);
 
+CREATE TABLE CHAIN3
+(
+  THREE       BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 30) PRIMARY KEY,
+  THREE_VALUE VARCHAR(20),
+  CHAIN4      BIGINT,
+  FOREIGN KEY (CHAIN4) REFERENCES CHAIN4 (FOUR)
+);
+
+CREATE TABLE CHAIN2
+(
+  TWO       BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 20) PRIMARY KEY,
+  TWO_VALUE VARCHAR(20),
+  CHAIN3    BIGINT,
+  FOREIGN KEY (CHAIN3) REFERENCES CHAIN3 (THREE)
+);
+
+CREATE TABLE CHAIN1
+(
+  ONE       BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 10) PRIMARY KEY,
+  ONE_VALUE VARCHAR(20),
+  CHAIN2    BIGINT,
+  FOREIGN KEY (CHAIN2) REFERENCES CHAIN2 (TWO)
+);
+
+CREATE TABLE CHAIN0
+(
+  ZERO       BIGINT GENERATED BY DEFAULT AS IDENTITY (START WITH 0) PRIMARY KEY,
+  ZERO_VALUE VARCHAR(20),
+  CHAIN1     BIGINT,
+  FOREIGN KEY (CHAIN1) REFERENCES CHAIN1 (ONE)
+);

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.core/JdbcAggregateTemplateIntegrationTests-mariadb.sql

@@ -1,13 +1,84 @@
-CREATE TABLE LEGO_SET ( id1 BIGINT AUTO_INCREMENT PRIMARY KEY, NAME VARCHAR(30));
-CREATE TABLE MANUAL ( id2 BIGINT AUTO_INCREMENT PRIMARY KEY, LEGO_SET BIGINT, ALTERNATIVE BIGINT, CONTENT VARCHAR(2000));
+CREATE TABLE LEGO_SET
+(
+  id1  BIGINT AUTO_INCREMENT PRIMARY KEY,
+  NAME VARCHAR(30)
+);
+CREATE TABLE MANUAL
+(
+  id2         BIGINT AUTO_INCREMENT PRIMARY KEY,
+  LEGO_SET    BIGINT,
+  ALTERNATIVE BIGINT,
+  CONTENT     VARCHAR(2000)
+);
 
-ALTER TABLE MANUAL ADD FOREIGN KEY (LEGO_SET)
-REFERENCES LEGO_SET(id1);
+ALTER TABLE MANUAL
+  ADD FOREIGN KEY (LEGO_SET)
+    REFERENCES LEGO_SET (id1);
 
-CREATE TABLE ONE_TO_ONE_PARENT ( id3 BIGINT AUTO_INCREMENT PRIMARY KEY, content VARCHAR(30));
-CREATE TABLE Child_No_Id (ONE_TO_ONE_PARENT INTEGER PRIMARY KEY, content VARCHAR(30));
+CREATE TABLE ONE_TO_ONE_PARENT
+(
+  id3     BIGINT AUTO_INCREMENT PRIMARY KEY,
+  content VARCHAR(30)
+);
+CREATE TABLE Child_No_Id
+(
+  ONE_TO_ONE_PARENT INTEGER PRIMARY KEY,
+  content           VARCHAR(30)
+);
 
-CREATE TABLE LIST_PARENT ( id4 BIGINT AUTO_INCREMENT PRIMARY KEY, NAME VARCHAR(100));
-CREATE TABLE element_no_id ( content VARCHAR(100), LIST_PARENT_key BIGINT, LIST_PARENT BIGINT);
+CREATE TABLE LIST_PARENT
+(
+  id4  BIGINT AUTO_INCREMENT PRIMARY KEY,
+  NAME VARCHAR(100)
+);
+CREATE TABLE element_no_id
+(
+  content         VARCHAR(100),
+  LIST_PARENT_key BIGINT,
+  LIST_PARENT     BIGINT
+);
 
-CREATE TABLE BYTE_ARRAY_OWNER (ID BIGINT AUTO_INCREMENT PRIMARY KEY, BINARY_DATA VARBINARY(20) NOT NULL)
+CREATE TABLE BYTE_ARRAY_OWNER
+(
+  ID          BIGINT AUTO_INCREMENT PRIMARY KEY,
+  BINARY_DATA VARBINARY(20) NOT NULL
+);
+
+CREATE TABLE CHAIN4
+(
+  FOUR       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  FOUR_VALUE VARCHAR(20)
+);
+
+
+CREATE TABLE CHAIN3
+(
+  THREE       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  THREE_VALUE VARCHAR(20),
+  CHAIN4      BIGINT,
+  FOREIGN KEY (CHAIN4) REFERENCES CHAIN4(FOUR)
+);
+
+CREATE TABLE CHAIN2
+(
+  TWO       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  TWO_VALUE VARCHAR(20),
+  CHAIN3      BIGINT,
+  FOREIGN KEY (CHAIN3) REFERENCES CHAIN3(THREE)
+);
+
+CREATE TABLE CHAIN1
+(
+  ONE       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  ONE_VALUE VARCHAR(20),
+  CHAIN2      BIGINT,
+  FOREIGN KEY (CHAIN2) REFERENCES CHAIN2(TWO)
+);
+
+CREATE TABLE CHAIN0
+(
+  ZERO       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  ZERO_VALUE VARCHAR(20),
+  CHAIN1      BIGINT,
+  FOREIGN KEY (CHAIN1) REFERENCES CHAIN1(ONE)
+);

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.core/JdbcAggregateTemplateIntegrationTests-mssql.sql

@@ -1,18 +1,93 @@
 DROP TABLE IF EXISTS MANUAL;
 DROP TABLE IF EXISTS LEGO_SET;
-CREATE TABLE LEGO_SET ( id1 BIGINT IDENTITY PRIMARY KEY, NAME VARCHAR(30));
-CREATE TABLE MANUAL ( id2 BIGINT IDENTITY PRIMARY KEY, LEGO_SET BIGINT, ALTERNATIVE BIGINT, CONTENT VARCHAR(2000));
-ALTER TABLE MANUAL ADD FOREIGN KEY (LEGO_SET) REFERENCES LEGO_SET(id1);
+CREATE TABLE LEGO_SET
+(
+  id1  BIGINT IDENTITY PRIMARY KEY,
+  NAME VARCHAR(30)
+);
+CREATE TABLE MANUAL
+(
+  id2         BIGINT IDENTITY PRIMARY KEY,
+  LEGO_SET    BIGINT,
+  ALTERNATIVE BIGINT,
+  CONTENT     VARCHAR(2000)
+);
+ALTER TABLE MANUAL
+  ADD FOREIGN KEY (LEGO_SET) REFERENCES LEGO_SET (id1);
 
 DROP TABLE IF EXISTS Child_No_Id;
 DROP TABLE IF EXISTS ONE_TO_ONE_PARENT;
-CREATE TABLE ONE_TO_ONE_PARENT ( id3 BIGINT IDENTITY PRIMARY KEY, content VARCHAR(30));
-CREATE TABLE Child_No_Id (ONE_TO_ONE_PARENT BIGINT PRIMARY KEY, content VARCHAR(30));
+CREATE TABLE ONE_TO_ONE_PARENT
+(
+  id3     BIGINT IDENTITY PRIMARY KEY,
+  content VARCHAR(30)
+);
+CREATE TABLE Child_No_Id
+(
+  ONE_TO_ONE_PARENT BIGINT PRIMARY KEY,
+  content           VARCHAR(30)
+);
 
 DROP TABLE IF EXISTS element_no_id;
 DROP TABLE IF EXISTS LIST_PARENT;
-CREATE TABLE LIST_PARENT ( id4 BIGINT IDENTITY  PRIMARY KEY, NAME VARCHAR(100));
-CREATE TABLE element_no_id ( content VARCHAR(100), LIST_PARENT_key BIGINT, LIST_PARENT BIGINT);
+CREATE TABLE LIST_PARENT
+(
+  id4  BIGINT IDENTITY PRIMARY KEY,
+  NAME VARCHAR(100)
+);
+CREATE TABLE element_no_id
+(
+  content         VARCHAR(100),
+  LIST_PARENT_key BIGINT,
+  LIST_PARENT     BIGINT
+);
 
 DROP TABLE IF EXISTS BYTE_ARRAY_OWNER;
-CREATE TABLE BYTE_ARRAY_OWNER (ID BIGINT IDENTITY  PRIMARY KEY, BINARY_DATA VARBINARY(20) NOT NULL)
+CREATE TABLE BYTE_ARRAY_OWNER
+(
+  ID          BIGINT IDENTITY PRIMARY KEY,
+  BINARY_DATA VARBINARY(20) NOT NULL
+);
+
+DROP TABLE IF EXISTS CHAIN4;
+CREATE TABLE CHAIN4
+(
+  FOUR       BIGINT IDENTITY PRIMARY KEY,
+  FOUR_VALUE VARCHAR(20)
+);
+
+DROP TABLE IF EXISTS CHAIN3;
+CREATE TABLE CHAIN3
+(
+  THREE       BIGINT IDENTITY PRIMARY KEY,
+  THREE_VALUE VARCHAR(20),
+  CHAIN4      BIGINT,
+  FOREIGN KEY (CHAIN4) REFERENCES CHAIN4 (FOUR)
+);
+
+DROP TABLE IF EXISTS CHAIN2;
+CREATE TABLE CHAIN2
+(
+  TWO       BIGINT IDENTITY PRIMARY KEY,
+  TWO_VALUE VARCHAR(20),
+  CHAIN3    BIGINT,
+  FOREIGN KEY (CHAIN3) REFERENCES CHAIN3 (THREE)
+);
+
+DROP TABLE IF EXISTS CHAIN1;
+CREATE TABLE CHAIN1
+(
+  ONE       BIGINT IDENTITY PRIMARY KEY,
+  ONE_VALUE VARCHAR(20),
+  CHAIN2    BIGINT,
+  FOREIGN KEY (CHAIN2) REFERENCES CHAIN2 (TWO)
+);
+
+DROP TABLE IF EXISTS CHAIN0;
+CREATE TABLE CHAIN0
+(
+  ZERO       BIGINT IDENTITY PRIMARY KEY,
+  ZERO_VALUE VARCHAR(20),
+  CHAIN1     BIGINT,
+  FOREIGN KEY (CHAIN1) REFERENCES CHAIN1 (ONE)
+);

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.core/JdbcAggregateTemplateIntegrationTests-mysql.sql

@@ -1,13 +1,84 @@
-CREATE TABLE LEGO_SET ( id1 BIGINT AUTO_INCREMENT PRIMARY KEY, NAME VARCHAR(30));
-CREATE TABLE MANUAL ( id2 BIGINT AUTO_INCREMENT PRIMARY KEY, LEGO_SET BIGINT, ALTERNATIVE BIGINT, CONTENT VARCHAR(2000));
+CREATE TABLE LEGO_SET
+(
+  id1  BIGINT AUTO_INCREMENT PRIMARY KEY,
+  NAME VARCHAR(30)
+);
+CREATE TABLE MANUAL
+(
+  id2         BIGINT AUTO_INCREMENT PRIMARY KEY,
+  LEGO_SET    BIGINT,
+  ALTERNATIVE BIGINT,
+  CONTENT     VARCHAR(2000)
+);
 
-ALTER TABLE MANUAL ADD FOREIGN KEY (LEGO_SET)
-REFERENCES LEGO_SET(id1);
+ALTER TABLE MANUAL
+  ADD FOREIGN KEY (LEGO_SET)
+    REFERENCES LEGO_SET (id1);
 
-CREATE TABLE ONE_TO_ONE_PARENT ( id3 BIGINT AUTO_INCREMENT PRIMARY KEY, content VARCHAR(30));
-CREATE TABLE Child_No_Id (ONE_TO_ONE_PARENT INTEGER PRIMARY KEY, content VARCHAR(30));
+CREATE TABLE ONE_TO_ONE_PARENT
+(
+  id3     BIGINT AUTO_INCREMENT PRIMARY KEY,
+  content VARCHAR(30)
+);
+CREATE TABLE Child_No_Id
+(
+  ONE_TO_ONE_PARENT INTEGER PRIMARY KEY,
+  content           VARCHAR(30)
+);
 
-CREATE TABLE LIST_PARENT ( id4 BIGINT AUTO_INCREMENT PRIMARY KEY, NAME VARCHAR(100));
-CREATE TABLE element_no_id ( content VARCHAR(100), LIST_PARENT_key BIGINT, LIST_PARENT BIGINT);
+CREATE TABLE LIST_PARENT
+(
+  id4  BIGINT AUTO_INCREMENT PRIMARY KEY,
+  NAME VARCHAR(100)
+);
+CREATE TABLE element_no_id
+(
+  content         VARCHAR(100),
+  LIST_PARENT_key BIGINT,
+  LIST_PARENT     BIGINT
+);
 
-CREATE TABLE BYTE_ARRAY_OWNER (ID BIGINT AUTO_INCREMENT PRIMARY KEY, BINARY_DATA VARBINARY(20) NOT NULL)
+CREATE TABLE BYTE_ARRAY_OWNER
+(
+  ID          BIGINT AUTO_INCREMENT PRIMARY KEY,
+  BINARY_DATA VARBINARY(20) NOT NULL
+);
+
+CREATE TABLE CHAIN4
+(
+  FOUR       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  FOUR_VALUE VARCHAR(20)
+);
+
+
+CREATE TABLE CHAIN3
+(
+  THREE       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  THREE_VALUE VARCHAR(20),
+  CHAIN4      BIGINT,
+  FOREIGN KEY (CHAIN4) REFERENCES CHAIN4(FOUR)
+);
+
+CREATE TABLE CHAIN2
+(
+  TWO       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  TWO_VALUE VARCHAR(20),
+  CHAIN3      BIGINT,
+  FOREIGN KEY (CHAIN3) REFERENCES CHAIN3(THREE)
+);
+
+CREATE TABLE CHAIN1
+(
+  ONE       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  ONE_VALUE VARCHAR(20),
+  CHAIN2      BIGINT,
+  FOREIGN KEY (CHAIN2) REFERENCES CHAIN2(TWO)
+);
+
+CREATE TABLE CHAIN0
+(
+  ZERO       BIGINT AUTO_INCREMENT PRIMARY KEY,
+  ZERO_VALUE VARCHAR(20),
+  CHAIN1      BIGINT,
+  FOREIGN KEY (CHAIN1) REFERENCES CHAIN1(ONE)
+);

spring-data-jdbc/src/test/resources/org.springframework.data.jdbc.core/JdbcAggregateTemplateIntegrationTests-postgres.sql

@@ -6,19 +6,99 @@ DROP TABLE LIST_PARENT;
 DROP TABLE element_no_id;
 DROP TABLE ARRAY_OWNER;
 DROP TABLE BYTE_ARRAY_OWNER;
+DROP TABLE CHAIN4;
+DROP TABLE CHAIN3;
+DROP TABLE CHAIN2;
+DROP TABLE CHAIN1;
+DROP TABLE CHAIN0;
 
-CREATE TABLE LEGO_SET ( id1 SERIAL PRIMARY KEY, NAME VARCHAR(30));
-CREATE TABLE MANUAL ( id2 SERIAL PRIMARY KEY, LEGO_SET BIGINT, ALTERNATIVE BIGINT, CONTENT VARCHAR(2000));
+CREATE TABLE LEGO_SET
+(
+  id1  SERIAL PRIMARY KEY,
+  NAME VARCHAR(30)
+);
+CREATE TABLE MANUAL
+(
+  id2         SERIAL PRIMARY KEY,
+  LEGO_SET    BIGINT,
+  ALTERNATIVE BIGINT,
+  CONTENT     VARCHAR(2000)
+);
 
-ALTER TABLE MANUAL ADD FOREIGN KEY (LEGO_SET)
-REFERENCES LEGO_SET(id1);
+ALTER TABLE MANUAL
+  ADD FOREIGN KEY (LEGO_SET)
+    REFERENCES LEGO_SET (id1);
 
-CREATE TABLE ONE_TO_ONE_PARENT ( id3 SERIAL PRIMARY KEY, content VARCHAR(30));
-CREATE TABLE Child_No_Id (ONE_TO_ONE_PARENT INTEGER PRIMARY KEY, content VARCHAR(30));
+CREATE TABLE ONE_TO_ONE_PARENT
+(
+  id3     SERIAL PRIMARY KEY,
+  content VARCHAR(30)
+);
+CREATE TABLE Child_No_Id
+(
+  ONE_TO_ONE_PARENT INTEGER PRIMARY KEY,
+  content           VARCHAR(30)
+);
 
-CREATE TABLE LIST_PARENT ( id4 SERIAL PRIMARY KEY, NAME VARCHAR(100));
-CREATE TABLE element_no_id ( content VARCHAR(100), LIST_PARENT_key BIGINT, LIST_PARENT INTEGER);
+CREATE TABLE LIST_PARENT
+(
+  id4  SERIAL PRIMARY KEY,
+  NAME VARCHAR(100)
+);
+CREATE TABLE element_no_id
+(
+  content         VARCHAR(100),
+  LIST_PARENT_key BIGINT,
+  LIST_PARENT     INTEGER
+);
 
-CREATE TABLE ARRAY_OWNER (ID SERIAL PRIMARY KEY, DIGITS VARCHAR(20)[10], MULTIDIMENSIONAL VARCHAR(20)[10][10]);
+CREATE TABLE ARRAY_OWNER
+(
+  ID               SERIAL PRIMARY KEY,
+  DIGITS           VARCHAR(20)[10],
+  MULTIDIMENSIONAL VARCHAR(20)[10][10]
+);
 
-CREATE TABLE BYTE_ARRAY_OWNER (ID SERIAL PRIMARY KEY, BINARY_DATA BYTEA NOT NULL)
+CREATE TABLE BYTE_ARRAY_OWNER
+(
+  ID          SERIAL PRIMARY KEY,
+  BINARY_DATA BYTEA NOT NULL
+);
+
+CREATE TABLE CHAIN4
+(
+  FOUR       SERIAL PRIMARY KEY,
+  FOUR_VALUE VARCHAR(20)
+);
+
+CREATE TABLE CHAIN3
+(
+  THREE       SERIAL PRIMARY KEY,
+  THREE_VALUE VARCHAR(20),
+  CHAIN4      BIGINT,
+  FOREIGN KEY (CHAIN4) REFERENCES CHAIN4 (FOUR)
+);
+
+CREATE TABLE CHAIN2
+(
+  TWO       SERIAL PRIMARY KEY,
+  TWO_VALUE VARCHAR(20),
+  CHAIN3    BIGINT,
+  FOREIGN KEY (CHAIN3) REFERENCES CHAIN3 (THREE)
+);
+
+CREATE TABLE CHAIN1
+(
+  ONE       SERIAL PRIMARY KEY,
+  ONE_VALUE VARCHAR(20),
+  CHAIN2    BIGINT,
+  FOREIGN KEY (CHAIN2) REFERENCES CHAIN2 (TWO)
+);
+
+CREATE TABLE CHAIN0
+(
+  ZERO       SERIAL PRIMARY KEY,
+  ZERO_VALUE VARCHAR(20),
+  CHAIN1     BIGINT,
+  FOREIGN KEY (CHAIN1) REFERENCES CHAIN1 (ONE)
+);

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/Column.java

@@ -24,6 +24,7 @@ import org.springframework.util.Assert;
  * Renders to: {@code <name>} or {@code <table(alias)>.<name>}.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  */
 public class Column extends AbstractSegment implements Expression, Named {
@@ -182,6 +183,16 @@ public class Column extends AbstractSegment implements Expression, Named {
 		return Conditions.in(this, expression);
 	}
 
+	/**
+	 * Creates a new {@link In} {@link Condition} given a subselects.
+	 *
+	 * @param subselect right side of the comparison.
+	 * @return the {@link In} condition.
+	 */
+	public In in(Select subselect) {
+		return Conditions.in(this, subselect);
+	}
+
 	/**
 	 * Creates a {@code IS NULL} condition.
 	 *

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/Conditions.java

@@ -139,7 +139,7 @@ public abstract class Conditions {
 	 * @param arg IN argument.
 	 * @return the {@link In} condition.
 	 */
-	public static Condition in(Expression columnOrExpression, Expression arg) {
+	public static In in(Expression columnOrExpression, Expression arg) {
 
 		Assert.notNull(columnOrExpression, "Comparison column or expression must not be null");
 		Assert.notNull(arg, "Expression argument must not be null");
@@ -184,7 +184,7 @@ public abstract class Conditions {
 	 * @param subselect the subselect.
 	 * @return the {@link In} condition.
 	 */
-	public static Condition in(Column column, Select subselect) {
+	public static In in(Column column, Select subselect) {
 
 		Assert.notNull(column, "Column must not be null");
 		Assert.notNull(subselect, "Subselect must not be null");

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/DefaultInsertBuilder.java

@@ -40,7 +40,7 @@ class DefaultInsertBuilder
 	 * @see org.springframework.data.relational.core.sql.InsertBuilder#into(org.springframework.data.relational.core.sql.Table)
 	 */
 	@Override
-	public InsertIntoColumnsAndValues into(Table table) {
+	public InsertIntoColumnsAndValuesWithBuild into(Table table) {
 
 		Assert.notNull(table, "Insert Into Table must not be null!");
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/DefaultSelectBuilder.java

@@ -30,6 +30,7 @@ import org.springframework.lang.Nullable;
  * Default {@link SelectBuilder} implementation.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  */
 class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAndJoin, SelectWhereAndOr {
@@ -249,6 +250,15 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 		return new JoinBuilder(table, this);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.sql.SelectBuilder.SelectJoin#join(org.springframework.data.relational.core.sql.Table)
+	 */
+	@Override
+	public SelectOn leftOuterJoin(Table table) {
+		return new JoinBuilder(table, this, JoinType.LEFT_OUTER_JOIN);
+	}
+
 	public DefaultSelectBuilder join(Join join) {
 		this.joins.add(join);
 
@@ -273,13 +283,21 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 
 		private final Table table;
 		private final DefaultSelectBuilder selectBuilder;
+		private final JoinType joinType;
 		private Expression from;
 		private Expression to;
 		private @Nullable Condition condition;
 
-		JoinBuilder(Table table, DefaultSelectBuilder selectBuilder) {
+
+		JoinBuilder(Table table, DefaultSelectBuilder selectBuilder, JoinType joinType) {
 			this.table = table;
 			this.selectBuilder = selectBuilder;
+
+			this.joinType = joinType;
+		}
+
+		JoinBuilder(Table table, DefaultSelectBuilder selectBuilder) {
+			this(table, selectBuilder, JoinType.JOIN);
 		}
 
 		/*
@@ -328,7 +346,7 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 		private Join finishJoin() {
 
 			finishCondition();
-			return new Join(JoinType.JOIN, table, condition);
+			return new Join(joinType, table, condition);
 		}
 
 		/*
@@ -391,6 +409,16 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 			return selectBuilder.join(table);
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.sql.SelectBuilder.SelectJoin#leftOuterJoin(org.springframework.data.relational.core.sql.Table)
+		 */
+		@Override
+		public SelectOn leftOuterJoin(Table table) {
+			selectBuilder.join(finishJoin());
+			return selectBuilder.leftOuterJoin(table);
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.relational.core.sql.SelectBuilder.SelectFromAndJoinCondition#limitOffset(long, long)

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/Expressions.java

@@ -19,6 +19,7 @@ package org.springframework.data.relational.core.sql;
  * Factory for common {@link Expression}s.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  * @see SQL
  * @see Conditions
@@ -55,7 +56,7 @@ public abstract class Expressions {
 	// Utility constructor.
 	private Expressions() {}
 
-	static class SimpleExpression extends AbstractSegment implements Expression {
+	static public class SimpleExpression extends AbstractSegment implements Expression {
 
 		private final String expression;
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/Functions.java

@@ -25,6 +25,7 @@ import org.springframework.util.Assert;
  * Factory for common {@link Expression function expressions}.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  * @see SQL
  * @see Expressions
@@ -38,7 +39,7 @@ public class Functions {
 	 * @param columns columns to apply count, must not be {@literal null}.
 	 * @return the new {@link SimpleFunction count function} for {@code columns}.
 	 */
-	public static SimpleFunction count(Column... columns) {
+	public static SimpleFunction count(Expression... columns) {
 
 		Assert.notNull(columns, "Columns must not be null!");
 		Assert.notEmpty(columns, "Columns must contains at least one column");

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/InsertBuilder.java

@@ -21,6 +21,7 @@ import java.util.Collection;
  * Entry point to construct an {@link Insert} statement.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  * @see StatementBuilder
  */
@@ -34,7 +35,7 @@ public interface InsertBuilder {
 	 * @see Into
 	 * @see SQL#table(String)
 	 */
-	InsertIntoColumnsAndValues into(Table table);
+	InsertIntoColumnsAndValuesWithBuild into(Table table);
 
 	/**
 	 * Interface exposing {@code WHERE} methods.

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/SelectBuilder.java

@@ -21,6 +21,7 @@ import java.util.Collection;
  * Entry point to construct a {@link Select} statement.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  * @see StatementBuilder
  */
@@ -472,6 +473,16 @@ public interface SelectBuilder {
 		 * @see SQL#table(String)
 		 */
 		SelectOn join(Table table);
+
+		/**
+		 * Declare a {@code LEFT OUTER JOIN} {@link Table}.
+		 *
+		 * @param table name of the table, must not be {@literal null}.
+		 * @return {@code this} builder.
+		 * @see Join
+		 * @see SQL#table(String)
+		 */
+		SelectOn leftOuterJoin(Table table);
 	}
 
 	/**

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/SelectValidator.java

@@ -26,6 +26,7 @@ import java.util.Stack;
  * {@code JOIN} clause.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  */
 class SelectValidator extends AbstractImportValidator {
@@ -93,18 +94,18 @@ class SelectValidator extends AbstractImportValidator {
 			return;
 		}
 
-		super.enter(segment);
+		if (segment instanceof Expression && parent instanceof Select) {
+			selectFieldCount++;
+		}
 
 		if (segment instanceof AsteriskFromTable && parent instanceof Select) {
 
 			Table table = ((AsteriskFromTable) segment).getTable();
 			requiredBySelect.add(table);
-			selectFieldCount++;
 		}
 
 		if (segment instanceof Column && (parent instanceof Select || parent instanceof SimpleFunction)) {
 
-			selectFieldCount++;
 			Table table = ((Column) segment).getTable();
 
 			if (table != null) {
@@ -124,6 +125,7 @@ class SelectValidator extends AbstractImportValidator {
 		if (segment instanceof Table && parent instanceof Join) {
 			join.add((Table) segment);
 		}
+		super.enter(segment);
 	}
 
 	/*

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/render/InsertStatementVisitor.java

@@ -25,6 +25,7 @@ import org.springframework.data.relational.core.sql.Visitable;
  * {@link PartRenderer} for {@link Insert} statements.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  * @since 1.1
  */
 class InsertStatementVisitor extends DelegatingVisitor implements PartRenderer {
@@ -94,17 +95,13 @@ class InsertStatementVisitor extends DelegatingVisitor implements PartRenderer {
 
 			builder.append("INSERT");
 
-			if (into.length() != 0) {
-				builder.append(" INTO ").append(into);
-			}
+			builder.append(" INTO ").append(into);
 
 			if (columns.length() != 0) {
 				builder.append(" (").append(columns).append(")");
 			}
 
-			if (values.length() != 0) {
-				builder.append(" VALUES(").append(values).append(")");
-			}
+			builder.append(" VALUES (").append(values).append(")");
 
 			return Delegation.leave();
 		}

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/render/SelectListVisitor.java

@@ -16,7 +16,9 @@
 package org.springframework.data.relational.core.sql.render;
 
 import org.springframework.data.relational.core.sql.Aliased;
+import org.springframework.data.relational.core.sql.AsteriskFromTable;
 import org.springframework.data.relational.core.sql.Column;
+import org.springframework.data.relational.core.sql.Expression;
 import org.springframework.data.relational.core.sql.SelectList;
 import org.springframework.data.relational.core.sql.SimpleFunction;
 import org.springframework.data.relational.core.sql.Table;
@@ -38,6 +40,7 @@ class SelectListVisitor extends TypedSubtreeVisitor<SelectList> implements PartR
 	private boolean insideFunction = false; // this is hackery and should be fix with a proper visitor for
 	// subelements.
 
+
 	SelectListVisitor(RenderContext context, RenderTarget target) {
 		this.context = context;
 		this.target = target;
@@ -57,8 +60,6 @@ class SelectListVisitor extends TypedSubtreeVisitor<SelectList> implements PartR
 		if (segment instanceof SimpleFunction) {
 			builder.append(((SimpleFunction) segment).getFunctionName()).append("(");
 			insideFunction = true;
-		} else {
-			insideFunction = false;
 		}
 
 		return super.enterNested(segment);
@@ -87,14 +88,26 @@ class SelectListVisitor extends TypedSubtreeVisitor<SelectList> implements PartR
 		}
 
 		if (segment instanceof SimpleFunction) {
+
 			builder.append(")");
+			if (segment instanceof Aliased) {
+				builder.append(" AS ").append(((Aliased) segment).getAlias());
+			}
+
+			insideFunction = false;
 			requiresComma = true;
 		} else if (segment instanceof Column) {
+
 			builder.append(context.getNamingStrategy().getName((Column) segment));
-			if (segment instanceof Aliased) {
+			if (segment instanceof Aliased && !insideFunction) {
 				builder.append(" AS ").append(((Aliased) segment).getAlias());
 			}
 			requiresComma = true;
+		} else if (segment instanceof AsteriskFromTable) {
+			// the toString of AsteriskFromTable includes the table name, which would cause it to appear twice.
+			builder.append("*");
+		} else if (segment instanceof Expression) {
+			builder.append(segment.toString());
 		}
 
 		return super.leaveNested(segment);

spring-data-relational/src/test/java/org/springframework/data/relational/core/sql/render/InsertRendererUnitTests.java

@@ -18,7 +18,6 @@ package org.springframework.data.relational.core.sql.render;
 import static org.assertj.core.api.Assertions.*;
 
 import org.junit.Test;
-
 import org.springframework.data.relational.core.sql.Insert;
 import org.springframework.data.relational.core.sql.SQL;
 import org.springframework.data.relational.core.sql.Table;
@@ -27,6 +26,7 @@ import org.springframework.data.relational.core.sql.Table;
  * Unit tests for {@link SqlRenderer}.
  *
  * @author Mark Paluch
+ * @author Jens Schauder
  */
 public class InsertRendererUnitTests {
 
@@ -37,7 +37,7 @@ public class InsertRendererUnitTests {
 
 		Insert insert = Insert.builder().into(bar).values(SQL.bindMarker()).build();
 
-		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar VALUES(?)");
+		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar VALUES (?)");
 	}
 
 	@Test // DATAJDBC-335
@@ -47,7 +47,7 @@ public class InsertRendererUnitTests {
 
 		Insert insert = Insert.builder().into(bar).column(bar.column("foo")).values(SQL.bindMarker()).build();
 
-		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar (foo) VALUES(?)");
+		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar (foo) VALUES (?)");
 	}
 
 	@Test // DATAJDBC-335
@@ -58,6 +58,17 @@ public class InsertRendererUnitTests {
 		Insert insert = Insert.builder().into(bar).columns(bar.columns("foo", "baz")).value(SQL.bindMarker())
 				.value(SQL.literalOf("foo")).build();
 
-		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar (foo, baz) VALUES(?, 'foo')");
+		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar (foo, baz) VALUES (?, 'foo')");
+	}
+
+	@Test // DATAJDBC-340
+	public void shouldRenderInsertWithZeroColumns() {
+
+		Table bar = SQL.table("bar");
+
+		Insert insert = Insert.builder().into(bar).build();
+
+		assertThat(SqlRenderer.toString(insert)).isEqualTo("INSERT INTO bar VALUES ()");
 	}
+
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/sql/render/SelectRendererUnitTests.java

@@ -17,10 +17,11 @@ package org.springframework.data.relational.core.sql.render;
 
 import static org.assertj.core.api.Assertions.*;
 
+import org.junit.Ignore;
 import org.junit.Test;
-
 import org.springframework.data.relational.core.sql.Column;
 import org.springframework.data.relational.core.sql.Conditions;
+import org.springframework.data.relational.core.sql.Expressions;
 import org.springframework.data.relational.core.sql.Functions;
 import org.springframework.data.relational.core.sql.OrderByField;
 import org.springframework.data.relational.core.sql.SQL;
@@ -93,6 +94,17 @@ public class SelectRendererUnitTests {
 		assertThat(SqlRenderer.toString(select)).isEqualTo("SELECT COUNT(bar.foo), bar.bar FROM bar");
 	}
 
+	@Test // DATAJDBC-340
+	public void shouldRenderCountFunctionWithAliasedColumn() {
+
+		Table table = SQL.table("bar");
+		Column foo = table.column("foo").as("foo_bar");
+
+		Select select = Select.builder().select(Functions.count(foo), foo).from(table).build();
+
+		assertThat(SqlRenderer.toString(select)).isEqualTo("SELECT COUNT(bar.foo), bar.foo AS foo_bar FROM bar");
+	}
+
 	@Test // DATAJDBC-309
 	public void shouldRenderSimpleJoin() {
 
@@ -107,6 +119,21 @@ public class SelectRendererUnitTests {
 				+ "JOIN department ON employee.department_id = department.id");
 	}
 
+	@Test // DATAJDBC-340
+	public void shouldRenderOuterJoin() {
+
+		Table employee = SQL.table("employee");
+		Table department = SQL.table("department");
+
+		Select select = Select.builder().select(employee.column("id"), department.column("name")) //
+				.from(employee) //
+				.leftOuterJoin(department).on(employee.column("department_id")).equals(department.column("id")) //
+				.build();
+
+		assertThat(SqlRenderer.toString(select)).isEqualTo("SELECT employee.id, department.name FROM employee "
+				+ "LEFT OUTER JOIN department ON employee.department_id = department.id");
+	}
+
 	@Test // DATAJDBC-309
 	public void shouldRenderSimpleJoinWithAnd() {
 
@@ -119,7 +146,7 @@ public class SelectRendererUnitTests {
 				.build();
 
 		assertThat(SqlRenderer.toString(select)).isEqualTo("SELECT employee.id, department.name FROM employee " //
-				+ "JOIN department ON employee.department_id = department.id "  //
+				+ "JOIN department ON employee.department_id = department.id " //
 				+ "AND employee.tenant = department.tenant");
 	}
 
@@ -246,7 +273,7 @@ public class SelectRendererUnitTests {
 
 		Select subselect = Select.builder().select(bah).from(floo).build();
 
-		Select select = Select.builder().select(bar).from(foo).where(Conditions.in(bar, subselect)).build();
+		Select select = Select.builder().select(bar).from(foo).where(bar.in(subselect)).build();
 
 		assertThat(SqlRenderer.toString(select))
 				.isEqualTo("SELECT foo.bar FROM foo WHERE foo.bar IN (SELECT floo.bah FROM floo)");
@@ -272,4 +299,44 @@ public class SelectRendererUnitTests {
 		assertThat(mapped).isEqualTo("SELECT foo.baR FROM foo WHERE foo.baR = foo.baZ");
 	}
 
+	@Test // DATAJDBC-340
+	public void shouldRenderCountStar() {
+
+		Select select = Select.builder() //
+				.select(Functions.count(Expressions.asterisk())) //
+				.from(SQL.table("foo")) //
+				.build();
+
+		String rendered = SqlRenderer.toString(select);
+
+		assertThat(rendered).isEqualTo("SELECT COUNT(*) FROM foo");
+	}
+
+	@Test // DATAJDBC-340
+	public void shouldRenderCountTableStar() {
+
+		Table foo = SQL.table("foo");
+		Select select = Select.builder() //
+				.select(Functions.count(foo.asterisk())) //
+				.from(foo) //
+				.build();
+
+		String rendered = SqlRenderer.toString(select);
+
+		assertThat(rendered).isEqualTo("SELECT COUNT(foo.*) FROM foo");
+	}
+
+	@Test // DATAJDBC-340
+	public void shouldRenderFunctionWithAlias() {
+
+		Table foo = SQL.table("foo");
+		Select select = Select.builder() //
+				.select(Functions.count(foo.asterisk()).as("counter")) //
+				.from(foo) //
+				.build();
+
+		String rendered = SqlRenderer.toString(select);
+
+		assertThat(rendered).isEqualTo("SELECT COUNT(foo.*) AS counter FROM foo");
+	}
 }