Add AssociationKey for bidirectionality detection

hibernate-core/src/main/java/org/hibernate/loader/ast/internal/LoaderSelectBuilder.java

@@ -8,6 +8,7 @@ package org.hibernate.loader.ast.internal;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -33,6 +34,7 @@ import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.JdbcMapping;
 import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
+import org.hibernate.metamodel.mapping.internal.EmbeddedAttributeMapping;
 import org.hibernate.metamodel.mapping.internal.SimpleForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.ordering.OrderByFragment;
 import org.hibernate.persister.collection.CollectionPersister;
@@ -80,6 +82,8 @@ import static org.hibernate.sql.ast.spi.SqlExpressionResolver.createColumnRefere
 public class LoaderSelectBuilder {
 	private static final Logger log = Logger.getLogger( LoaderSelectBuilder.class );
 
+	private HashMap<String, List<Fetch>> visitedNavigablePath = new HashMap<>();
+
 	/**
 	 * Create an SQL AST select-statement based on matching one-or-more keys
 	 *
@@ -209,7 +213,7 @@ public class LoaderSelectBuilder {
 				new SimpleFromClauseAccessImpl(),
 				lockOptions,
 				this::visitFetches,
-				numberOfKeysToLoad > 1,
+				numberOfKeysToLoad > 1 || restrictedPart instanceof ForeignKeyDescriptor,
 				creationContext
 		);
 
@@ -430,15 +434,22 @@ public class LoaderSelectBuilder {
 		}
 
 		final List<Fetch> fetches = new ArrayList<>();
+		String fullPath = fetchParent.getNavigablePath().getFullPath();
+		final List<Fetch> fullPathFetches = visitedNavigablePath.get( fullPath );
+		if ( fullPathFetches != null ) {
+			return fullPathFetches;
+		}
 
 		final BiConsumer<Fetchable, Boolean> processor = createFetchableBiConsumer( fetchParent, querySpec, creationState, fetches );
 
 		final FetchableContainer referencedMappingContainer = fetchParent.getReferencedMappingContainer();
 		if ( fetchParent.getNavigablePath().getParent() != null ) {
-			referencedMappingContainer.visitKeyFetchables( fetchable -> processor.accept( fetchable, true ), null );
+			referencedMappingContainer.visitKeyFetchables(
+					fetchable -> processor.accept( fetchable, true ), null );
 		}
-		referencedMappingContainer.visitFetchables( fetchable -> processor.accept( fetchable, false ), null );
+		referencedMappingContainer.visitFetchables(
-
+				fetchable -> processor.accept( fetchable, false ), null );
+		visitedNavigablePath.put( fullPath, fetches );
 		return fetches;
 	}
 
@@ -501,12 +512,14 @@ public class LoaderSelectBuilder {
 					joined = false;
 				}
 				else if ( fetchDepth > maximumFetchDepth ) {
-					return;
+					if ( !( fetchable instanceof BasicValuedModelPart ) && !( fetchable instanceof EmbeddedAttributeMapping ) ) {
+						return;
+					}
 				}
 			}
 
 			try {
-				if ( !( fetchable instanceof BasicValuedModelPart ) ) {
+				if ( !( fetchable instanceof BasicValuedModelPart ) && !( fetchable instanceof EmbeddedAttributeMapping ) ) {
 					fetchDepth++;
 				}
 				final Fetch fetch = fetchable.generateFetch(
@@ -537,7 +550,7 @@ public class LoaderSelectBuilder {
 				}
 			}
 			finally {
-				if ( !( fetchable instanceof BasicValuedModelPart ) ) {
+				if ( !( fetchable instanceof BasicValuedModelPart ) && !( fetchable instanceof EmbeddedAttributeMapping ) ) {
 					fetchDepth--;
 				}
 				if ( entityGraphTraversalState != null ) {

hibernate-core/src/main/java/org/hibernate/loader/ast/internal/LoaderSqlAstCreationState.java

@@ -7,7 +7,9 @@
 package org.hibernate.loader.ast.internal;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import javax.persistence.CacheRetrieveMode;
 import javax.persistence.CacheStoreMode;
 
@@ -16,6 +18,7 @@ import org.hibernate.LockMode;
 import org.hibernate.LockOptions;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.graph.spi.AppliedGraph;
+import org.hibernate.metamodel.mapping.AssociationKey;
 import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.query.Limit;
 import org.hibernate.query.NavigablePath;
@@ -48,13 +51,15 @@ public class LoaderSqlAstCreationState
 
 	private final QuerySpec querySpec;
 	private final SqlAliasBaseManager sqlAliasBaseManager;
-	private boolean forceIdentifierSelection;
+	private final boolean forceIdentifierSelection;
 	private final SqlAstCreationContext sf;
 	private final SqlAstQuerySpecProcessingStateImpl processingState;
 	private final FromClauseAccess fromClauseAccess;
 	private final LockOptions lockOptions;
 	private final FetchProcessor fetchProcessor;
 
+	private Set<AssociationKey> visitedAssociationKeys = new HashSet<>();
+
 	public LoaderSqlAstCreationState(
 			QuerySpec querySpec,
 			SqlAliasBaseManager sqlAliasBaseManager,
@@ -143,6 +148,16 @@ public class LoaderSqlAstCreationState
 		return this;
 	}
 
+	@Override
+	public void registerVisitedAssociationKey(AssociationKey associationKey) {
+		visitedAssociationKeys.add( associationKey );
+	}
+
+	@Override
+	public boolean isAssociationKeyVisited(AssociationKey associationKey) {
+		return visitedAssociationKeys.contains( associationKey );
+	}
+
 	@Override
 	public ModelPart resolveModelPart(NavigablePath navigablePath) {
 		// for now, let's assume that the navigable-path refers to TableGroup

hibernate-core/src/main/java/org/hibernate/loader/ast/internal/SingleUniqueKeyEntityLoaderStandard.java

@@ -20,7 +20,9 @@ import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.loader.ast.spi.SingleUniqueKeyEntityLoader;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.metamodel.mapping.JdbcMapping;
+import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.SingularAttributeMapping;
+import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
 import org.hibernate.query.spi.QueryOptions;
 import org.hibernate.query.spi.QueryOptionsAdapter;
 import org.hibernate.query.spi.QueryParameterBindings;
@@ -40,13 +42,18 @@ import org.hibernate.sql.exec.spi.JdbcSelect;
  */
 public class SingleUniqueKeyEntityLoaderStandard<T> implements SingleUniqueKeyEntityLoader<T> {
 	private final EntityMappingType entityDescriptor;
-	private final SingularAttributeMapping uniqueKeyAttribute;
+	private final ModelPart uniqueKeyAttribute;
 
 	public SingleUniqueKeyEntityLoaderStandard(
 			EntityMappingType entityDescriptor,
 			SingularAttributeMapping uniqueKeyAttribute) {
 		this.entityDescriptor = entityDescriptor;
-		this.uniqueKeyAttribute = uniqueKeyAttribute;
+		if ( uniqueKeyAttribute instanceof ToOneAttributeMapping ) {
+			this.uniqueKeyAttribute = ( (ToOneAttributeMapping) uniqueKeyAttribute ).getForeignKeyDescriptor();
+		}
+		else {
+			this.uniqueKeyAttribute = uniqueKeyAttribute;
+		}
 	}
 
 	@Override
@@ -142,7 +149,11 @@ public class SingleUniqueKeyEntityLoaderStandard<T> implements SingleUniqueKeyEn
 				true
 		);
 
-		assert list.size() == 1;
+		int size = list.size();
+		assert size <= 1;
+		if ( size == 0 ) {
+			return null;
+		}
 
 		//noinspection unchecked
 		return (T) list.get( 0 );

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/AssociationKey.java

@@ -0,0 +1,51 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.metamodel.mapping;
+
+import java.util.List;
+
+/**
+ * @author Andrea Boriero
+ */
+public class AssociationKey {
+	private final String table;
+	private final List<String> columns;
+
+	public AssociationKey(String table, List<String> columns) {
+		this.table = table;
+		this.columns = columns;
+	}
+
+	@Override
+	public boolean equals(Object o) {
+		if ( this == o ) {
+			return true;
+		}
+		if ( o == null || getClass() != o.getClass() ) {
+			return false;
+		}
+
+		final AssociationKey that = (AssociationKey) o;
+		return table.equals( that.table ) && columns.equals( that.columns );
+	}
+
+	@Override
+	public int hashCode() {
+		return table.hashCode();
+	}
+
+	private String str;
+
+	@Override
+	public String toString() {
+		if ( str == null ) {
+			str = "AssociationKey(table=" + table + ", columns={" + String.join( ",", columns ) + "})";
+		}
+		return str;
+	}
+
+}

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/EmbeddableMappingType.java

@@ -345,7 +345,18 @@ public class EmbeddableMappingType implements ManagedMappingType {
 			Clause clause,
 			TypeConfiguration typeConfiguration) {
 		attributeMappings.forEach(
-				(s, attributeMapping) -> attributeMapping.visitJdbcTypes( action, clause, typeConfiguration )
+				(s, attributeMapping) -> {
+					if ( attributeMapping instanceof ToOneAttributeMapping ) {
+						( (ToOneAttributeMapping) attributeMapping ).getKeyTargetMatchPart().visitJdbcTypes(
+								action,
+								clause,
+								typeConfiguration
+						);
+					}
+					else {
+						attributeMapping.visitJdbcTypes( action, clause, typeConfiguration );
+					}
+				}
 		);
 	}
 

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/EntityAssociationMapping.java

@@ -6,14 +6,12 @@
  */
 package org.hibernate.metamodel.mapping;
 
-import org.hibernate.sql.results.graph.Fetchable;
-
 /**
  * Commonality between `many-to-one`, `one-to-one` and `any`, as well as entity-valued collection elements and map-keys
  *
  * @author Steve Ebersole
  */
-public interface EntityAssociationMapping extends ModelPart, Fetchable {
+public interface EntityAssociationMapping extends ModelPart, Association {
 	@Override
 	default String getFetchableName() {
 		return getPartName();

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/EntityMappingType.java

@@ -202,7 +202,6 @@ public interface EntityMappingType extends ManagedMappingType, Loadable {
 						final Object value = assembler == null ? UNFETCHED_PROPERTY : assembler.assemble( rowProcessingState );
 
 						values[index++] = value;
-
 					}
 				}
 		);

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/ForeignKeyDescriptor.java

@@ -51,10 +51,6 @@ public interface ForeignKeyDescriptor extends VirtualModelPart {
 		return PART_NAME;
 	}
 
-	String getReferringTableExpression();
-
-	String getTargetTableExpression();
-
 	/**
 	 * Visits the FK "referring" columns
 	 */
@@ -67,6 +63,5 @@ public interface ForeignKeyDescriptor extends VirtualModelPart {
 
 	void visitTargetColumns(ColumnConsumer consumer);
 
-
+	AssociationKey getAssociationKey();
-	boolean areTargetColumnNamesEqualsTo(String[] columnNames);
 }

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/PluralAttributeMapping.java

@@ -55,4 +55,6 @@ public interface PluralAttributeMapping
 	}
 
 	String getSeparateCollectionTable();
+
+	String getMappedBy();
 }

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/AbstractSingularAttributeMapping.java

@@ -7,7 +7,6 @@
 package org.hibernate.metamodel.mapping.internal;
 
 import org.hibernate.engine.FetchStrategy;
-import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.ManagedMappingType;
 import org.hibernate.metamodel.mapping.SingularAttributeMapping;
 import org.hibernate.metamodel.mapping.StateArrayContributorMetadataAccess;

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/EmbeddedAttributeMapping.java

@@ -98,6 +98,7 @@ public class EmbeddedAttributeMapping
 	public EmbeddableMappingType getEmbeddableTypeDescriptor() {
 		return embeddableMappingType;
 	}
+
 	@Override
 	public SingularAttributeMapping getParentInjectionAttributeMapping() {
 		// todo (6.0) : implement

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/EmbeddedForeignKeyDescriptor.java

@@ -8,8 +8,10 @@ package org.hibernate.metamodel.mapping.internal;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 
 import org.hibernate.HibernateException;
+import org.hibernate.metamodel.mapping.AssociationKey;
 import org.hibernate.metamodel.mapping.ColumnConsumer;
 import org.hibernate.metamodel.mapping.EmbeddableValuedModelPart;
 import org.hibernate.metamodel.mapping.EntityMappingType;
@@ -20,6 +22,7 @@ import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.model.domain.NavigableRole;
 import org.hibernate.query.ComparisonOperator;
 import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.ast.Clause;
 import org.hibernate.sql.ast.SqlAstJoinType;
 import org.hibernate.sql.ast.spi.SqlAstCreationContext;
 import org.hibernate.sql.ast.spi.SqlAstCreationState;
@@ -49,6 +52,7 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor, Model
 	private final List<String> targetColumnExpressions;
 	private final EmbeddableValuedModelPart mappingType;
 	private final List<JdbcMapping> jdbcMappings;
+	private AssociationKey associationKey;
 
 	public EmbeddedForeignKeyDescriptor(
 			EmbeddedIdentifierMappingImpl mappingType,
@@ -305,16 +309,6 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor, Model
 		throw new IllegalStateException( "Could not resolve binding for table `" + table + "`" );
 	}
 
-	@Override
-	public String getReferringTableExpression() {
-		return keyColumnContainingTable;
-	}
-
-	@Override
-	public String getTargetTableExpression() {
-		return targetColumnContainingTable;
-	}
-
 	@Override
 	public void visitReferringColumns(ColumnConsumer consumer) {
 		for ( int i = 0; i < keyColumnExpressions.size(); i++ ) {
@@ -330,17 +324,11 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor, Model
 	}
 
 	@Override
-	public boolean areTargetColumnNamesEqualsTo(String[] columnNames) {
+	public AssociationKey getAssociationKey() {
-		int length = columnNames.length;
+		if ( associationKey == null ) {
-		if ( length != targetColumnExpressions.size() ) {
+			associationKey = new AssociationKey( keyColumnContainingTable, keyColumnExpressions );
-			return false;
 		}
-		for ( int i = 0; i < length; i++ ) {
+		return associationKey;
-			if ( !targetColumnExpressions.contains( columnNames[i] ) ) {
-				return false;
-			}
-		}
-		return true;
 	}
 
 	@Override
@@ -363,4 +351,9 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor, Model
 		throw new UnsupportedOperationException();
 	}
 
+	@Override
+	public void visitJdbcTypes(
+			Consumer<JdbcMapping> action, Clause clause, TypeConfiguration typeConfiguration) {
+		mappingType.visitJdbcTypes( action, clause, typeConfiguration );
+	}
 }

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/EntityCollectionPart.java

@@ -22,6 +22,7 @@ import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.model.domain.NavigableRole;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.ast.spi.FromClauseAccess;
 import org.hibernate.sql.ast.tree.from.TableGroup;
 import org.hibernate.sql.results.graph.DomainResult;
 import org.hibernate.sql.results.graph.DomainResultCreationState;
@@ -38,7 +39,7 @@ import org.hibernate.type.descriptor.java.JavaTypeDescriptor;
  * @author Steve Ebersole
  */
 public class EntityCollectionPart
-		implements CollectionPart, EntityAssociationMapping, EntityValuedFetchable, Association, FetchOptions {
+		implements CollectionPart, EntityAssociationMapping, EntityValuedFetchable, FetchOptions {
 	private final NavigableRole navigableRole;
 	private final CollectionPersister collectionDescriptor;
 	private final Nature nature;
@@ -137,18 +138,17 @@ public class EntityCollectionPart
 			LockMode lockMode,
 			String resultVariable,
 			DomainResultCreationState creationState) {
-//		assert fetchParent.getReferencedMappingContainer() instanceof PluralAttributeMapping;
-
 		// find or create the TableGroup associated with this `fetchablePath`
-		creationState.getSqlAstCreationState().getFromClauseAccess().resolveTableGroup(
+		final FromClauseAccess fromClauseAccess = creationState.getSqlAstCreationState().getFromClauseAccess();
+		creationState.registerVisitedAssociationKey( getForeignKeyDescriptor().getAssociationKey() );
+
+		fromClauseAccess.resolveTableGroup(
 				fetchablePath,
 				np -> {
 					// We need to create one.  The Result will be able to find it later by path
 
 					// first, find the collection's TableGroup
-					final TableGroup collectionTableGroup = creationState.getSqlAstCreationState()
+					final TableGroup collectionTableGroup = fromClauseAccess.getTableGroup( fetchParent.getNavigablePath() );
-							.getFromClauseAccess()
-							.getTableGroup( fetchParent.getNavigablePath() );
 
 					assert collectionTableGroup != null;
 

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/PluralAttributeMappingImpl.java

@@ -22,6 +22,7 @@ import org.hibernate.jpa.spi.JpaCompliance;
 import org.hibernate.mapping.Collection;
 import org.hibernate.mapping.IndexedCollection;
 import org.hibernate.mapping.List;
+import org.hibernate.mapping.OneToOne;
 import org.hibernate.mapping.Property;
 import org.hibernate.mapping.Value;
 import org.hibernate.metamodel.mapping.BasicValuedModelPart;
@@ -45,6 +46,7 @@ import org.hibernate.persister.entity.Joinable;
 import org.hibernate.property.access.spi.PropertyAccess;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.ast.SqlAstJoinType;
+import org.hibernate.sql.ast.spi.FromClauseAccess;
 import org.hibernate.sql.ast.spi.SqlAliasBase;
 import org.hibernate.sql.ast.spi.SqlAliasBaseGenerator;
 import org.hibernate.sql.ast.spi.SqlAliasStemHelper;
@@ -94,6 +96,7 @@ public class PluralAttributeMappingImpl extends AbstractAttributeMapping impleme
 	private final FetchStyle fetchStyle;
 
 	private final CascadeStyle cascadeStyle;
+	private final String mappedBy;
 
 	private final CollectionPersister collectionDescriptor;
 	private final String separateCollectionTable;
@@ -169,6 +172,7 @@ public class PluralAttributeMappingImpl extends AbstractAttributeMapping impleme
 		this.fetchStyle = fetchStyle;
 		this.cascadeStyle = cascadeStyle;
 		this.collectionDescriptor = collectionDescriptor;
+		this.mappedBy = bootDescriptor.getMappedByProperty();
 
 		this.sqlAliasStem = SqlAliasStemHelper.INSTANCE.generateStemFromAttributeName( attributeName );
 
@@ -380,6 +384,11 @@ public class PluralAttributeMappingImpl extends AbstractAttributeMapping impleme
 		return separateCollectionTable;
 	}
 
+	@Override
+	public String getMappedBy() {
+		return mappedBy;
+	}
+
 	@Override
 	public int getStateArrayPosition() {
 		return stateArrayPosition;
@@ -442,12 +451,15 @@ public class PluralAttributeMappingImpl extends AbstractAttributeMapping impleme
 			DomainResultCreationState creationState) {
 		final SqlAstCreationState sqlAstCreationState = creationState.getSqlAstCreationState();
 
+		creationState.registerVisitedAssociationKey( fkDescriptor.getAssociationKey() );
+
 		if ( fetchTiming == FetchTiming.IMMEDIATE) {
 			if ( selected ) {
-				final TableGroup collectionTableGroup = sqlAstCreationState.getFromClauseAccess().resolveTableGroup(
+				final FromClauseAccess fromClauseAccess = sqlAstCreationState.getFromClauseAccess();
+				final TableGroup collectionTableGroup = fromClauseAccess.resolveTableGroup(
 						fetchablePath,
 						p -> {
-							final TableGroup lhsTableGroup = sqlAstCreationState.getFromClauseAccess().getTableGroup(
+							final TableGroup lhsTableGroup = fromClauseAccess.getTableGroup(
 									fetchParent.getNavigablePath() );
 							final TableGroupJoin tableGroupJoin = createTableGroupJoin(
 									fetchablePath,

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/SimpleForeignKeyDescriptor.java

@@ -6,6 +6,7 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Consumer;
@@ -14,6 +15,7 @@ import org.hibernate.LockMode;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.AssociationKey;
 import org.hibernate.metamodel.mapping.BasicValuedModelPart;
 import org.hibernate.metamodel.mapping.ColumnConsumer;
 import org.hibernate.metamodel.mapping.EntityMappingType;
@@ -53,6 +55,7 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 	private final String targetColumnContainingTable;
 	private final String targetColumnExpression;
 	private final JdbcMapping jdbcMapping;
+	private AssociationKey associationKey;
 
 	public SimpleForeignKeyDescriptor(
 			String keyColumnContainingTable,
@@ -285,32 +288,23 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 		throw new UnsupportedOperationException();
 	}
 
-	@Override
-	public String getReferringTableExpression() {
-		return keyColumnContainingTable;
-	}
-
 	@Override
 	public void visitReferringColumns(ColumnConsumer consumer) {
 		consumer.accept( keyColumnContainingTable, keyColumnExpression, jdbcMapping );
 	}
 
-	@Override
-	public String getTargetTableExpression() {
-		return targetColumnContainingTable;
-	}
-
 	@Override
 	public void visitTargetColumns(ColumnConsumer consumer) {
 		consumer.accept( targetColumnContainingTable, targetColumnExpression, jdbcMapping );
 	}
 
 	@Override
-	public boolean areTargetColumnNamesEqualsTo(String[] columnNames) {
+	public AssociationKey getAssociationKey() {
-		if ( columnNames.length != 1 ) {
+		if ( associationKey == null ) {
-			return false;
+			final List<String> associationKeyColumns = Collections.singletonList( keyColumnExpression );
+			associationKey = new AssociationKey( keyColumnContainingTable, associationKeyColumns );
 		}
-		return targetColumnExpression.equals( columnNames[0] );
+		return associationKey;
 	}
 
 	@Override
@@ -393,11 +387,9 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 		return jdbcMapping;
 	}
 
-	public String getTargetColumnContainingTable() {
+	@Override
-		return targetColumnContainingTable;
+	public String toString() {
-	}
+		return "SimpleForeignKeyDescriptor : " + keyColumnContainingTable + "." + keyColumnExpression
-
+				+ " --> " + targetColumnContainingTable + "." + targetColumnExpression;
-	public String getTargetColumnExpression() {
-		return targetColumnExpression;
 	}
 }

hibernate-core/src/main/java/org/hibernate/metamodel/mapping/internal/ToOneAttributeMapping.java

@@ -6,23 +6,24 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
-import org.hibernate.HibernateException;
 import org.hibernate.LockMode;
 import org.hibernate.engine.FetchStrategy;
 import org.hibernate.engine.FetchTiming;
+import org.hibernate.internal.util.StringHelper;
 import org.hibernate.mapping.ManyToOne;
 import org.hibernate.mapping.OneToOne;
 import org.hibernate.mapping.ToOne;
-import org.hibernate.metamodel.mapping.Association;
+import org.hibernate.metamodel.mapping.AssociationKey;
 import org.hibernate.metamodel.mapping.EntityAssociationMapping;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.ManagedMappingType;
 import org.hibernate.metamodel.mapping.ModelPart;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.metamodel.mapping.StateArrayContributorMetadataAccess;
 import org.hibernate.metamodel.model.domain.NavigableRole;
 import org.hibernate.persister.entity.EntityPersister;
-import org.hibernate.persister.entity.SingleTableEntityPersister;
 import org.hibernate.property.access.spi.PropertyAccess;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.ast.SqlAstJoinType;
@@ -42,21 +43,21 @@ import org.hibernate.sql.results.graph.DomainResult;
 import org.hibernate.sql.results.graph.DomainResultCreationState;
 import org.hibernate.sql.results.graph.Fetch;
 import org.hibernate.sql.results.graph.FetchParent;
-import org.hibernate.sql.results.graph.Fetchable;
+import org.hibernate.sql.results.graph.embeddable.EmbeddableValuedFetchable;
 import org.hibernate.sql.results.graph.entity.EntityFetch;
-import org.hibernate.sql.results.graph.entity.EntityResultGraphNode;
 import org.hibernate.sql.results.graph.entity.EntityValuedFetchable;
-import org.hibernate.sql.results.graph.entity.internal.EntityFetchDelayedImpl;
+import org.hibernate.sql.results.graph.entity.internal.EntityDelayedFetchImpl;
 import org.hibernate.sql.results.graph.entity.internal.EntityFetchJoinedImpl;
 import org.hibernate.sql.results.graph.entity.internal.EntityFetchSelectImpl;
-import org.hibernate.sql.results.internal.domain.BiDirectionalFetchImpl;
+import org.hibernate.sql.results.internal.domain.CircularFetchImpl;
+import org.hibernate.sql.results.internal.domain.CircularBiDirectionalFetchImpl;
 import org.hibernate.type.ForeignKeyDirection;
 
 /**
  * @author Steve Ebersole
  */
 public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
-		implements EntityValuedFetchable, EntityAssociationMapping, Association, TableGroupJoinProducer {
+		implements EntityValuedFetchable, EntityAssociationMapping, TableGroupJoinProducer {
 
 	public enum Cardinality {
 		ONE_TO_ONE,
@@ -75,6 +76,7 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 	private final boolean referringPrimaryKey;
 
 	private final Cardinality cardinality;
+	private String mappedBy;
 
 	private ForeignKeyDescriptor foreignKeyDescriptor;
 	private ForeignKeyDirection foreignKeyDirection;
@@ -124,6 +126,13 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 		else {
 			assert bootValue instanceof OneToOne;
 			cardinality = Cardinality.ONE_TO_ONE;
+			String mappedByProperty = ( (OneToOne) bootValue ).getMappedByProperty();
+			if ( mappedByProperty == null ) {
+				mappedBy = StringHelper.nullIfEmpty( referencedPropertyName );
+			}
+			else {
+				mappedBy = mappedByProperty;
+			}
 		}
 
 		this.navigableRole = navigableRole;
@@ -169,101 +178,142 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 			NavigablePath fetchablePath,
 			FetchParent fetchParent,
 			DomainResultCreationState creationState) {
-		// NOTE - a circular fetch reference ultimately needs 2 pieces of information:
+		final AssociationKey associationKey = foreignKeyDescriptor.getAssociationKey();
-		//		1) The NavigablePath that is circular (`fetchablePath`)
-		//		2) The NavigablePath to the entity-valued-reference that is the "other side" of the circularity
 
-		final ModelPart parentModelPart = fetchParent.getReferencedModePart();
+		if ( creationState.isAssociationKeyVisited( associationKey ) ) {
-
+			NavigablePath parent = fetchablePath.getParent();
-		if ( ! Fetchable.class.isInstance( parentModelPart ) ) {
+			ModelPart modelPart = creationState.resolveModelPart( parent );
-			// the `fetchParent` would have to be a Fetch as well for this to be circular...
+			if ( modelPart instanceof EmbeddedIdentifierMappingImpl ) {
-			return null;
+				while ( parent.getFullPath().endsWith( EntityIdentifierMapping.ROLE_LOCAL_NAME ) ) {
-		}
+					parent = parent.getParent();
-
-		final FetchParent associationFetchParent = fetchParent.resolveContainingAssociationParent();
-		if ( associationFetchParent == null ) {
-			return null;
-		}
-		final ModelPart referencedModePart = associationFetchParent.getReferencedModePart();
-		assert referencedModePart instanceof Association;
-
-		final Association associationParent = (Association) referencedModePart;
-
-		if ( foreignKeyDescriptor.equals( associationParent.getForeignKeyDescriptor() ) ) {
-			// we need to determine the NavigablePath referring to the entity that the bi-dir
-			// fetch will "return" for its Assembler.  so we walk "up" the FetchParent graph
-			// to find the "referenced entity" reference
-
-			return createBiDirectionalFetch( fetchablePath, fetchParent );
-		}
-
-		// this is the case of a JoinTable
-		// 	PARENT(id)
-		// 	PARENT_CHILD(parent_id, child_id)
-		// 	CHILD(id)
-		// 	the FKDescriptor for the association `Parent.child` will be
-		//		PARENT_CHILD.child.id -> CHILD.id
-		// and the FKDescriptor for the association `Child.parent` will be
-		//		PARENT_CHILD.parent.id -> PARENT.id
-		// in such a case the associationParent.getIdentifyingColumnExpressions() is PARENT_CHILD.parent_id
-		// while the getIdentifyingColumnExpressions for this association is PARENT_CHILD.child_id
-		// so we will check if the parentAssociation ForeignKey Target match with the association entity identifier table and columns
-		final ForeignKeyDescriptor associationParentForeignKeyDescriptor = associationParent.getForeignKeyDescriptor();
-		if ( referencedModePart instanceof ToOneAttributeMapping
-				&& ( (ToOneAttributeMapping) referencedModePart ).getDeclaringType() == getPartMappingType() ) {
-			if ( this.foreignKeyDescriptor.getReferringTableExpression()
-					.equals( associationParentForeignKeyDescriptor.getReferringTableExpression() ) ) {
-				final SingleTableEntityPersister entityPersister = (SingleTableEntityPersister) getDeclaringType();
-				if ( associationParentForeignKeyDescriptor.getTargetTableExpression()
-						.equals( entityPersister.getTableName() ) ) {
-					final String[] identifierColumnNames = entityPersister.getIdentifierColumnNames();
-					if ( associationParentForeignKeyDescriptor.areTargetColumnNamesEqualsTo( identifierColumnNames ) ) {
-						return createBiDirectionalFetch( fetchablePath, fetchParent );
-					}
-					return null;
 				}
+			}
+			while ( modelPart instanceof EmbeddableValuedFetchable ) {
+				parent = parent.getParent();
+				modelPart = creationState.resolveModelPart( parent );
+			}
 
+			if ( this.mappedBy != null && parent.getFullPath().endsWith( this.mappedBy )  ) {
+				/*
+					class Child {
+						@OneToOne(mappedBy = "biologicalChild")
+						private Mother mother;
+					}
+
+					class Mother {
+						@OneToOne
+						private Child biologicalChild;
+					}
+
+					fetchablePath= Mother.biologicalChild.mother
+					this.mappedBy = "biologicalChild"
+					parent.getFullPath() = "Mother.biologicalChild"
+				 */
+				return createCircularBiDirectionalFetch(
+						fetchablePath,
+						fetchParent,
+						parent.getParent(),
+						LockMode.READ
+				);
+			}
+
+			/*
+				check if mappedBy is on the other side of the association
+			 */
+			final String otherSideMappedBy = getOtherSideMappedBy( modelPart, parent.getParent(), creationState );
+			if ( otherSideMappedBy != null ) {
+					/*
+						class Child {
+							@OneToOne(mappedBy = "biologicalChild")
+							private Mother mother;
+						}
+
+						class Mother {
+							@OneToOne
+							private Child biologicalChild;
+						}
+
+						fetchablePath = "Child.mother.biologicalChild"
+						otherSideAssociationModelPart = ToOneAttributeMapping("Child.mother")
+						otherSideMappedBy = "biologicalChild"
+
+					 */
+
+				if ( fetchablePath.getFullPath().endsWith( otherSideMappedBy ) ) {
+					return createCircularBiDirectionalFetch(
+							fetchablePath,
+							fetchParent,
+							parent.getParent(),
+							LockMode.READ
+					);
+				}
+			}
+			/*
+						class Child {
+							@OneToOne
+							private Mother mother;
+						}
+
+						class Mother {
+							@OneToOne
+							private Child stepMother;
+						}
+
+				We have a cirularity but it is not bidirectional
+			 */
+			if ( referringPrimaryKey ) {
+				final TableGroup parentTableGroup = creationState
+						.getSqlAstCreationState()
+						.getFromClauseAccess()
+						.getTableGroup( fetchParent.getNavigablePath() );
+				return new CircularFetchImpl(
+						getEntityMappingType(),
+						getTiming(),
+						fetchablePath,
+						fetchParent,
+						this,
+						fetchablePath,
+						foreignKeyDescriptor.createDomainResult( fetchablePath, parentTableGroup, creationState )
+				);
 			}
 		}
 		return null;
 	}
 
-	private Fetch createBiDirectionalFetch(NavigablePath fetchablePath, FetchParent fetchParent) {
+	private String getOtherSideMappedBy(
-		final EntityResultGraphNode referencedEntityReference = resolveEntityGraphNode( fetchParent );
+			ModelPart modelPart,
-
+			NavigablePath parentOfParent,
-		if ( referencedEntityReference == null ) {
+			DomainResultCreationState creationState) {
-			throw new HibernateException(
+		if ( modelPart instanceof ToOneAttributeMapping ) {
-					"Could not locate entity-valued reference for circular path `" + fetchablePath + "`"
+			return ( (ToOneAttributeMapping) modelPart ).getMappedBy();
-			);
 		}
 
-		return new BiDirectionalFetchImpl(
+		if ( modelPart instanceof EntityCollectionPart ) {
+			return ( (PluralAttributeMapping) creationState.resolveModelPart( parentOfParent ) ).getMappedBy();
+		}
+
+		return null;
+	}
+
+	public String getMappedBy(){
+		return mappedBy;
+	}
+
+	private Fetch createCircularBiDirectionalFetch(
+			NavigablePath fetchablePath,
+			FetchParent fetchParent,
+			NavigablePath referencedNavigablePath,
+			LockMode lockMode) {
+		return new CircularBiDirectionalFetchImpl(
 				FetchTiming.IMMEDIATE,
 				fetchablePath,
 				fetchParent,
 				this,
-				referencedEntityReference.getNavigablePath()
+				lockMode,
+				referencedNavigablePath
 		);
 	}
 
-	protected EntityResultGraphNode resolveEntityGraphNode(FetchParent fetchParent) {
-		FetchParent processingParent = fetchParent;
-		while ( processingParent != null ) {
-			if ( processingParent instanceof EntityResultGraphNode ) {
-				return (EntityResultGraphNode) processingParent;
-			}
-
-			if ( processingParent instanceof Fetch ) {
-				processingParent = ( (Fetch) processingParent ).getFetchParent();
-				continue;
-			}
-
-			processingParent = null;
-		}
-
-		return null;
-	}
-
 	@Override
 	public EntityFetch generateFetch(
 			FetchParent fetchParent,
@@ -273,6 +323,7 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 			LockMode lockMode,
 			String resultVariable,
 			DomainResultCreationState creationState) {
+
 		final SqlAstCreationState sqlAstCreationState = creationState.getSqlAstCreationState();
 		final FromClauseAccess fromClauseAccess = sqlAstCreationState.getFromClauseAccess();
 
@@ -292,7 +343,6 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 							sqlAstJoinType = SqlAstJoinType.INNER;
 						}
 
-
 						final TableGroupJoin tableGroupJoin = createTableGroupJoin(
 								fetchablePath,
 								parentTableGroup,
@@ -306,6 +356,7 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 					}
 			);
 
+			creationState.registerVisitedAssociationKey( foreignKeyDescriptor.getAssociationKey() );
 			return new EntityFetchJoinedImpl(
 					fetchParent,
 					this,
@@ -319,12 +370,41 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 		//noinspection rawtypes
 		final DomainResult keyResult;
 
+		/*
+			1. No JoinTable
+				Model:
+					EntityA{
+						@ManyToOne
+						EntityB b
+					}
+
+					EntityB{
+						@ManyToOne
+						EntityA a
+					}
+
+				Relational:
+					ENTITY_A( id )
+					ENTITY_B( id, entity_a_id)
+
+				1.1 EntityA -> EntityB : as keyResult we need ENTITY_B.id
+				1.2 EntityB -> EntityA : as keyResult we need ENTITY_B.entity_a_id (FK referring column)
+
+			2. JoinTable
+
+		 */
+
+		boolean selectByUniqueKey;
 		if ( referringPrimaryKey ) {
+			// case 1.2
 			keyResult = foreignKeyDescriptor.createDomainResult( fetchablePath, parentTableGroup, creationState );
+			selectByUniqueKey = false;
 		}
 		else {
-			keyResult = ( (EntityPersister) getDeclaringType() ).getIdentifierMapping()
+			keyResult = ((EntityPersister) getDeclaringType()).getIdentifierMapping()
 					.createDomainResult( fetchablePath, parentTableGroup, null, creationState );
+			// case 1.1
+			selectByUniqueKey = true;
 		}
 
 		assert !selected;
@@ -332,19 +412,17 @@ public class ToOneAttributeMapping extends AbstractSingularAttributeMapping
 			return new EntityFetchSelectImpl(
 					fetchParent,
 					this,
-					lockMode,
 					isNullable,
 					fetchablePath,
 					keyResult,
+					selectByUniqueKey,
 					creationState
 			);
 		}
 
-		return new EntityFetchDelayedImpl(
+		return new EntityDelayedFetchImpl(
 				fetchParent,
 				this,
-				lockMode,
-				isNullable,
 				fetchablePath,
 				keyResult
 		);

hibernate-core/src/main/java/org/hibernate/persister/entity/AbstractEntityPersister.java

@@ -6347,10 +6347,10 @@ public abstract class AbstractEntityPersister
 	public void visitKeyFetchables(
 			Consumer<Fetchable> fetchableConsumer,
 			EntityMappingType treatTargetType) {
-		if ( getIdentifierMapping() instanceof FetchableContainer ) {
+//		if ( getIdentifierMapping() instanceof FetchableContainer ) {
-			// essentially means the entity has a composite id - ask the embeddable to visit its fetchables
+//			// essentially means the entity has a composite id - ask the embeddable to visit its fetchables
-			( (FetchableContainer) getIdentifierMapping() ).visitFetchables( fetchableConsumer, treatTargetType );
+//			( (FetchableContainer) getIdentifierMapping() ).visitFetchables( fetchableConsumer, treatTargetType );
-		}
+//		}
 		// otherwise, nothing to do
 	}
 

hibernate-core/src/main/java/org/hibernate/persister/entity/UniqueKeyLoadable.java

@@ -18,6 +18,7 @@ public interface UniqueKeyLoadable extends Loadable {
 	 */
 	Object loadByUniqueKey(String propertyName, Object uniqueKey, SharedSessionContractImplementor session);
 
+
 	/**
 	 * Get the property number of the unique key property
 	 */

hibernate-core/src/main/java/org/hibernate/sql/results/graph/DomainResultCreationState.java

@@ -10,6 +10,7 @@ import java.util.List;
 
 import org.hibernate.LockMode;
 import org.hibernate.NotYetImplementedFor6Exception;
+import org.hibernate.metamodel.mapping.AssociationKey;
 import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.results.jdbc.spi.JdbcValues;
@@ -31,6 +32,13 @@ public interface DomainResultCreationState {
 		return (SqlAliasBaseManager) getSqlAstCreationState().getSqlAliasBaseGenerator();
 	}
 
+	default void registerVisitedAssociationKey(AssociationKey associationKey){
+	}
+
+	default boolean isAssociationKeyVisited(AssociationKey associationKey){
+		return false;
+	}
+
 	/**
 	 * Resolve the ModelPart associated with a given NavigablePath.  More specific ModelParts should be preferred - e.g.
 	 * the SingularAssociationAttributeMapping rather than just the EntityTypeMapping for the associated type

hibernate-core/src/main/java/org/hibernate/sql/results/graph/embeddable/internal/EmbeddableForeignKeyResultImpl.java

@@ -29,7 +29,7 @@ import org.hibernate.sql.results.graph.basic.BasicFetch;
 import org.hibernate.sql.results.graph.basic.BasicResult;
 import org.hibernate.sql.results.graph.embeddable.EmbeddableInitializer;
 import org.hibernate.sql.results.graph.embeddable.EmbeddableResultGraphNode;
-import org.hibernate.sql.results.graph.entity.internal.EntityFetchDelayedImpl;
+import org.hibernate.sql.results.graph.entity.internal.EntityDelayedFetchImpl;
 import org.hibernate.sql.results.graph.entity.internal.EntityFetchSelectImpl;
 
 /**
@@ -76,11 +76,9 @@ public class EmbeddableForeignKeyResultImpl<T>
 			);
 			Fetch fetch;
 			if ( toOneAttributeMapping.getMappedFetchOptions().getTiming() == FetchTiming.DELAYED ) {
-				fetch = new EntityFetchDelayedImpl(
+				fetch = new EntityDelayedFetchImpl(
 						this,
 						toOneAttributeMapping,
-						null,
-						false,
 						navigablePath.append( fetchable.getFetchableName() ),
 						domainResult
 				);
@@ -89,10 +87,10 @@ public class EmbeddableForeignKeyResultImpl<T>
 				fetch = new EntityFetchSelectImpl(
 						this,
 						toOneAttributeMapping,
-						null,
 						false,
 						navigablePath.append( fetchable.getFetchableName() ),
 						domainResult,
+						false,
 						creationState
 				);
 			}
@@ -129,6 +127,11 @@ public class EmbeddableForeignKeyResultImpl<T>
 		return new EmbeddableAssembler( initializer );
 	}
 
+	@Override
+	public NavigablePath getNavigablePath() {
+		return super.getNavigablePath().append( "{fk}");
+	}
+
 	@Override
 	public EmbeddableMappingType getReferencedMappingType() {
 		return (EmbeddableMappingType) getFetchContainer().getPartMappingType();

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/AbstractEntityInitializer.java

@@ -121,7 +121,6 @@ public abstract class AbstractEntityInitializer extends AbstractFetchParentAcces
 							for ( int i = 0; i < identifierInitializers.size(); i++ ) {
 								final Initializer existing = identifierInitializers.get( i );
 								if ( existing.getNavigablePath().equals( navigablePath ) ) {
-									identifierInitializers.add( existing );
 									return existing;
 								}
 							}
@@ -328,9 +327,24 @@ public abstract class AbstractEntityInitializer extends AbstractFetchParentAcces
 
 	@SuppressWarnings("WeakerAccess")
 	protected void initializeIdentifier(RowProcessingState rowProcessingState) {
+		if ( EntityLoadingLogger.TRACE_ENABLED ) {
+			EntityLoadingLogger.LOGGER.tracef(
+					"(%s) Beginning Initializer#initializeIdentifier process for entity (%s) ",
+					StringHelper.collapse( this.getClass().getName() ),
+					getNavigablePath()
+			);
+		}
+
 		identifierInitializers.forEach( initializer -> initializer.resolveKey( rowProcessingState ) );
 		identifierInitializers.forEach( initializer -> initializer.resolveInstance( rowProcessingState ) );
-		identifierInitializers.forEach( initializer -> initializer.initializeInstance( rowProcessingState ) );
+
+		if ( EntityLoadingLogger.TRACE_ENABLED ) {
+			EntityLoadingLogger.LOGGER.tracef(
+					"(%s) Fiish Initializer#initializeIdentifier process for entity (%s) ",
+					StringHelper.collapse( this.getClass().getName() ),
+					getNavigablePath()
+			);
+		}
 	}
 
 	@SuppressWarnings("WeakerAccess")
@@ -351,8 +365,7 @@ public abstract class AbstractEntityInitializer extends AbstractFetchParentAcces
 		else {
 			id = identifierAssembler.assemble(
 					rowProcessingState,
-					jdbcValuesSourceProcessingState
+					jdbcValuesSourceProcessingState.getProcessingOptions()
-							.getProcessingOptions()
 			);
 		}
 
@@ -364,6 +377,10 @@ public abstract class AbstractEntityInitializer extends AbstractFetchParentAcces
 		//		2) build the EntityKey
 		this.entityKey = new EntityKey( id, concreteDescriptor );
 
+		if ( jdbcValuesSourceProcessingState.findInitializer( entityKey ) == null ) {
+			jdbcValuesSourceProcessingState.registerInitilaizer( entityKey, this );
+		}
+
 		//		3) schedule the EntityKey for batch loading, if possible
 		if ( concreteDescriptor.isBatchLoadable() ) {
 			if ( !session.getPersistenceContext().containsEntity( entityKey ) ) {
@@ -377,6 +394,7 @@ public abstract class AbstractEntityInitializer extends AbstractFetchParentAcces
 		if ( missing ) {
 			return;
 		}
+		identifierInitializers.forEach( initializer -> initializer.initializeInstance( rowProcessingState ) );
 
 		final Object entityIdentifier = entityKey.getIdentifier();
 
@@ -509,7 +527,6 @@ public abstract class AbstractEntityInitializer extends AbstractFetchParentAcces
 			);
 		}
 
-
 		// todo (6.0): do we really need this check ?
 		if ( persistenceContext.containsEntity( entityKey ) ) {
 			Status status = persistenceContext.getEntry( persistenceContext.getEntity( entityKey ) )

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/EntityInitializer.java

@@ -6,6 +6,7 @@
  */
 package org.hibernate.sql.results.graph.entity;
 
+import org.hibernate.LockMode;
 import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.sql.results.graph.FetchParentAccess;
 import org.hibernate.sql.results.graph.Initializer;

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/EntityDelayedFetchImpl.java

@@ -6,7 +6,6 @@
  */
 package org.hibernate.sql.results.graph.entity.internal;
 
-import org.hibernate.LockMode;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
 import org.hibernate.query.NavigablePath;
@@ -21,23 +20,16 @@ import org.hibernate.sql.results.graph.entity.EntityInitializer;
  * @author Andrea Boriero
  * @author Steve Ebersole
  */
-public class EntityFetchDelayedImpl extends AbstractNonJoinedEntityFetch {
+public class EntityDelayedFetchImpl extends AbstractNonJoinedEntityFetch {
-	private final LockMode lockMode;
-	private final boolean nullable;
 
 	private final DomainResult keyResult;
 
-	public EntityFetchDelayedImpl(
+	public EntityDelayedFetchImpl(
 			FetchParent fetchParent,
 			ToOneAttributeMapping fetchedAttribute,
-			LockMode lockMode,
-			boolean nullable,
 			NavigablePath navigablePath,
 			DomainResult keyResult) {
 		super( navigablePath, fetchedAttribute, fetchParent );
-		this.lockMode = lockMode;
-		this.nullable = nullable;
-
 		this.keyResult = keyResult;
 	}
 
@@ -57,7 +49,7 @@ public class EntityFetchDelayedImpl extends AbstractNonJoinedEntityFetch {
 			AssemblerCreationState creationState) {
 		final EntityInitializer entityInitializer = (EntityInitializer) creationState.resolveInitializer(
 				getNavigablePath(),
-				() -> new EntityFetchDelayedInitializer(
+				() -> new EntityDelayedFetchInitializer(
 						getNavigablePath(),
 						getEntityValuedModelPart().getEntityMappingType().getEntityPersister(),
 						keyResult.createResultAssembler( creationState )

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/EntityDelayedFetchInitializer.java

@@ -11,6 +11,7 @@ import java.util.function.Consumer;
 import org.hibernate.NotYetImplementedFor6Exception;
 import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.engine.spi.PersistenceContext;
+import org.hibernate.internal.log.LoggingHelper;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.results.graph.AbstractFetchParentAccess;
@@ -23,7 +24,7 @@ import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
  * @author Andrea Boriero
  * @author Steve Ebersole
  */
-public class EntityFetchDelayedInitializer extends AbstractFetchParentAccess implements EntityInitializer {
+public class EntityDelayedFetchInitializer extends AbstractFetchParentAccess implements EntityInitializer {
 
 	private final NavigablePath navigablePath;
 	private final EntityPersister concreteDescriptor;
@@ -32,7 +33,7 @@ public class EntityFetchDelayedInitializer extends AbstractFetchParentAccess imp
 	private Object entityInstance;
 	private Object identifier;
 
-	protected EntityFetchDelayedInitializer(
+	public EntityDelayedFetchInitializer(
 			NavigablePath fetchedNavigable,
 			EntityPersister concreteDescriptor,
 			DomainResultAssembler identifierAssembler) {
@@ -147,4 +148,9 @@ public class EntityFetchDelayedInitializer extends AbstractFetchParentAccess imp
 		}
 	}
 
+	@Override
+	public String toString() {
+		return "EntityDelayedFetchInitializer(" + LoggingHelper.toLoggableString( navigablePath ) + ")";
+	}
+
 }

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/EntityFetchSelectImpl.java

@@ -6,9 +6,9 @@
  */
 package org.hibernate.sql.results.graph.entity.internal;
 
-import org.hibernate.LockMode;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
+import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.results.graph.AssemblerCreationState;
 import org.hibernate.sql.results.graph.DomainResult;
@@ -26,18 +26,20 @@ import org.hibernate.sql.results.graph.entity.EntityInitializer;
 public class EntityFetchSelectImpl extends AbstractNonJoinedEntityFetch {
 	private final boolean nullable;
 	private final DomainResult result;
+	private final boolean selectByUniqueKey;
 
 	public EntityFetchSelectImpl(
 			FetchParent fetchParent,
 			ToOneAttributeMapping fetchedAttribute,
-			LockMode lockMode,
 			boolean nullable,
 			NavigablePath navigablePath,
 			DomainResult result,
+			boolean selectByUniqueKey,
 			DomainResultCreationState creationState) {
 		super( navigablePath, fetchedAttribute, fetchParent );
 		this.nullable = nullable;
 		this.result = result;
+		this.selectByUniqueKey = selectByUniqueKey;
 	}
 
 	@Override
@@ -54,12 +56,26 @@ public class EntityFetchSelectImpl extends AbstractNonJoinedEntityFetch {
 	public DomainResultAssembler createAssembler(FetchParentAccess parentAccess, AssemblerCreationState creationState) {
 		final EntityInitializer initializer = (EntityInitializer) creationState.resolveInitializer(
 				getNavigablePath(),
-				() -> new EntitySelectFetchInitializer(
+				() -> {
-						getNavigablePath(),
+
-						getReferencedMappingContainer().getEntityPersister(),
+					EntityPersister entityPersister = getReferencedMappingContainer().getEntityPersister();
-						result.createResultAssembler( creationState ),
+
-						nullable
+					if ( selectByUniqueKey ) {
-				)
+						return new EntitySelectFetchByUniqueKeyInitializer(
+								(ToOneAttributeMapping) getFetchedMapping(),
+								getNavigablePath(),
+								entityPersister,
+								result.createResultAssembler( creationState ),
+								nullable
+						);
+					}
+					return new EntitySelectFetchInitializer(
+							getNavigablePath(),
+							entityPersister,
+							result.createResultAssembler( creationState ),
+							nullable
+					);
+				}
 		);
 
 		return new EntityAssembler( getResultJavaTypeDescriptor(), initializer );

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/EntityResultInitializer.java

@@ -55,6 +55,6 @@ public class EntityResultInitializer extends AbstractEntityInitializer {
 
 	@Override
 	public String toString() {
-		return "EntityRootInitializer(" + getNavigablePath().getFullPath() + ")";
+		return CONCRETE_NAME + "(" + getNavigablePath().getFullPath() + ")";
 	}
 }

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/EntitySelectFetchByUniqueKeyInitializer.java

@@ -0,0 +1,77 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.sql.results.graph.entity.internal;
+
+import org.hibernate.engine.spi.EntityUniqueKey;
+import org.hibernate.engine.spi.PersistenceContext;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
+import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.persister.entity.UniqueKeyLoadable;
+import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.results.graph.DomainResultAssembler;
+import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
+
+/**
+ * @author Andrea Boriero
+ */
+public class EntitySelectFetchByUniqueKeyInitializer extends EntitySelectFetchInitializer {
+	private final ToOneAttributeMapping fetchedAttribute;
+
+	public EntitySelectFetchByUniqueKeyInitializer(
+			ToOneAttributeMapping fetchedAttribute,
+			NavigablePath fetchedNavigable,
+			EntityPersister concreteDescriptor,
+			DomainResultAssembler identifierAssembler,
+			boolean nullable) {
+		super( fetchedNavigable, concreteDescriptor, identifierAssembler, nullable );
+		this.fetchedAttribute = fetchedAttribute;
+	}
+
+	@Override
+	public void initializeInstance(RowProcessingState rowProcessingState) {
+		if ( entityInstance != null ) {
+			return;
+		}
+
+		final Object entityIdentifier = identifierAssembler.assemble( rowProcessingState );
+		if ( entityIdentifier == null ) {
+			return;
+		}
+		final String entityName = concreteDescriptor.getEntityName();
+		String uniqueKeyPropertyName = fetchedAttribute.getMappedBy();
+
+		final SharedSessionContractImplementor session = rowProcessingState.getSession();
+
+		EntityUniqueKey euk = new EntityUniqueKey(
+				entityName,
+				uniqueKeyPropertyName,
+				entityIdentifier,
+				concreteDescriptor.getIdentifierType(),
+				concreteDescriptor.getEntityMode(),
+				session.getFactory()
+		);
+		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
+		entityInstance = persistenceContext.getEntity( euk );
+		if ( entityInstance == null ) {
+			entityInstance = ( (UniqueKeyLoadable) concreteDescriptor ).loadByUniqueKey(
+					uniqueKeyPropertyName,
+					entityIdentifier,
+					session
+			);
+
+			// If the entity was not in the Persistence Context, but was found now,
+			// add it to the Persistence Context
+			if ( entityInstance != null ) {
+				persistenceContext.addEntity( euk, entityInstance );
+			}
+		}
+		if ( entityInstance != null ) {
+			entityInstance = persistenceContext.proxyFor( entityInstance );
+		}
+	}
+}

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/EntitySelectFetchInitializer.java

@@ -10,28 +10,39 @@ import java.util.function.Consumer;
 
 import org.hibernate.NotYetImplementedFor6Exception;
 import org.hibernate.engine.spi.EntityKey;
+import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.internal.log.LoggingHelper;
+import org.hibernate.internal.util.StringHelper;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.proxy.HibernateProxy;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.results.graph.AbstractFetchParentAccess;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
+import org.hibernate.sql.results.graph.Initializer;
 import org.hibernate.sql.results.graph.entity.EntityInitializer;
+import org.hibernate.sql.results.graph.entity.EntityLoadingLogger;
+import org.hibernate.sql.results.graph.entity.EntityValuedFetchable;
+import org.hibernate.sql.results.graph.entity.LoadingEntityEntry;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 
+import static org.hibernate.internal.log.LoggingHelper.toLoggableString;
+
 /**
  * @author Andrea Boriero
  */
 public class EntitySelectFetchInitializer extends AbstractFetchParentAccess implements EntityInitializer {
+	private static final String CONCRETE_NAME = EntitySelectFetchInitializer.class.getSimpleName();
+
 	private final NavigablePath navigablePath;
-	private final EntityPersister concreteDescriptor;
-	private final DomainResultAssembler identifierAssembler;
 	private final boolean isEnhancedForLazyLoading;
 	private final boolean nullable;
 
-	private Object entityInstance;
+	protected final EntityPersister concreteDescriptor;
+	protected final DomainResultAssembler identifierAssembler;
+	protected Object entityInstance;
 
-	protected EntitySelectFetchInitializer(
+	public EntitySelectFetchInitializer(
 			NavigablePath fetchedNavigable,
 			EntityPersister concreteDescriptor,
 			DomainResultAssembler identifierAssembler,
@@ -55,7 +66,6 @@ public class EntitySelectFetchInitializer extends AbstractFetchParentAccess impl
 
 	@Override
 	public void resolveInstance(RowProcessingState rowProcessingState) {
-
 	}
 
 	@Override
@@ -64,21 +74,106 @@ public class EntitySelectFetchInitializer extends AbstractFetchParentAccess impl
 			return;
 		}
 
-		final Object id = identifierAssembler.assemble( rowProcessingState );
+		final Object entityIdentifier = identifierAssembler.assemble( rowProcessingState );
-		if ( id == null ) {
+
+		if ( entityIdentifier == null ) {
 			return;
 		}
 
-		final String entityName = concreteDescriptor.getEntityName();
+		if ( EntityLoadingLogger.TRACE_ENABLED ) {
+			EntityLoadingLogger.LOGGER.tracef(
+					"(%s) Beginning Initializer#resolveInstance process for entity (%s) : %s",
+					StringHelper.collapse( this.getClass().getName() ),
+					getNavigablePath(),
+					entityIdentifier
+			);
+		}
 		final SharedSessionContractImplementor session = rowProcessingState.getSession();
+		final String entityName = concreteDescriptor.getEntityName();
 
+		final EntityKey entityKey = new EntityKey( entityIdentifier, concreteDescriptor );
+
+		Initializer initializer = rowProcessingState.getJdbcValuesSourceProcessingState().findInitializer(
+				entityKey );
+
+		if ( initializer != null ) {
+			if ( EntityLoadingLogger.DEBUG_ENABLED ) {
+				EntityLoadingLogger.LOGGER.debugf(
+						"(%s) Found an initializer for entity (%s) : %s",
+						CONCRETE_NAME,
+						toLoggableString( getNavigablePath(), entityIdentifier ),
+						entityIdentifier
+				);
+			}
+			initializer.resolveInstance( rowProcessingState );
+			entityInstance = initializer.getInitializedInstance();
+			return;
+		}
+
+		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
+		entityInstance = persistenceContext.getEntity( entityKey );
+		if ( entityInstance != null ) {
+			return;
+		}
+
+		final LoadingEntityEntry existingLoadingEntry = session
+				.getPersistenceContext()
+				.getLoadContexts()
+				.findLoadingEntityEntry( entityKey );
+
+		if ( existingLoadingEntry != null ) {
+			if ( EntityLoadingLogger.DEBUG_ENABLED ) {
+				EntityLoadingLogger.LOGGER.debugf(
+						"(%s) Found existing loading entry [%s] - using loading instance",
+						CONCRETE_NAME,
+						toLoggableString(
+								getNavigablePath(),
+								entityIdentifier
+						)
+				);
+			}
+			this.entityInstance = existingLoadingEntry.getEntityInstance();
+
+			if ( existingLoadingEntry.getEntityInitializer() != this ) {
+				// the entity is already being loaded elsewhere
+				if ( EntityLoadingLogger.DEBUG_ENABLED ) {
+					EntityLoadingLogger.LOGGER.debugf(
+							"(%s) Entity [%s] being loaded by another initializer [%s] - skipping processing",
+							CONCRETE_NAME,
+							toLoggableString( getNavigablePath(), entityIdentifier ),
+							existingLoadingEntry.getEntityInitializer()
+					);
+				}
+
+				// EARLY EXIT!!!
+				return;
+			}
+		}
+
+		if ( EntityLoadingLogger.DEBUG_ENABLED ) {
+			EntityLoadingLogger.LOGGER.debugf(
+					"(%s) Invoking session#internalLoad for entity (%s) : %s",
+					CONCRETE_NAME,
+					toLoggableString( getNavigablePath(), entityIdentifier ),
+					entityIdentifier
+			);
+		}
 		entityInstance = session.internalLoad(
 				entityName,
-				id,
+				entityIdentifier,
 				true,
 				nullable
 		);
 
+		if ( EntityLoadingLogger.DEBUG_ENABLED ) {
+			EntityLoadingLogger.LOGGER.debugf(
+					"(%s) Entity [%s] : %s has being loaded by session.internalLoad.",
+					CONCRETE_NAME,
+					toLoggableString( getNavigablePath(), entityIdentifier ),
+					entityIdentifier
+			);
+		}
+
 		if ( entityInstance instanceof HibernateProxy && isEnhancedForLazyLoading ) {
 			( (HibernateProxy) entityInstance ).getHibernateLazyInitializer().setUnwrap( true );
 		}
@@ -120,4 +215,9 @@ public class EntitySelectFetchInitializer extends AbstractFetchParentAccess impl
 			super.registerResolutionListener( listener );
 		}
 	}
+
+	@Override
+	public String toString() {
+		return "EntitySelectFetchInitializer(" + LoggingHelper.toLoggableString( getNavigablePath() ) + ")";
+	}
 }

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/internal/RootEntityResultImpl.java

@@ -0,0 +1,213 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.sql.results.graph.entity.internal;
+
+import java.util.ArrayList;
+
+import org.hibernate.LockMode;
+import org.hibernate.metamodel.mapping.EntityDiscriminatorMapping;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
+import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.EntityValuedModelPart;
+import org.hibernate.metamodel.mapping.EntityVersionMapping;
+import org.hibernate.metamodel.mapping.ManagedMappingType;
+import org.hibernate.metamodel.mapping.internal.SingleAttributeIdentifierMapping;
+import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
+import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.ast.tree.from.TableGroup;
+import org.hibernate.sql.results.graph.AbstractFetchParent;
+import org.hibernate.sql.results.graph.AssemblerCreationState;
+import org.hibernate.sql.results.graph.DomainResult;
+import org.hibernate.sql.results.graph.DomainResultAssembler;
+import org.hibernate.sql.results.graph.DomainResultCreationState;
+import org.hibernate.sql.results.graph.FetchableContainer;
+import org.hibernate.sql.results.graph.entity.EntityInitializer;
+import org.hibernate.sql.results.graph.entity.EntityResult;
+import org.hibernate.sql.results.graph.entity.EntityResultGraphNode;
+import org.hibernate.type.descriptor.java.JavaTypeDescriptor;
+
+/**
+ * @author Andrea Boriero
+ */
+public class RootEntityResultImpl extends AbstractFetchParent implements EntityResultGraphNode, EntityResult {
+
+	private final String resultVariable;
+
+	private final EntityValuedModelPart referencedModelPart;
+	private final DomainResult discriminatorResult;
+	private final DomainResult versionResult;
+	private DomainResult identifierResult;
+	private final LockMode lockMode;
+
+	public RootEntityResultImpl(
+			NavigablePath navigablePath,
+			EntityValuedModelPart entityValuedModelPart,
+			String resultVariable,
+			DomainResultCreationState creationState) {
+		this( navigablePath, entityValuedModelPart, resultVariable, null, creationState );
+	}
+
+	@SuppressWarnings("WeakerAccess")
+	public RootEntityResultImpl(
+			NavigablePath navigablePath,
+			EntityValuedModelPart entityValuedModelPart,
+			String resultVariable,
+			EntityMappingType targetType,
+			DomainResultCreationState creationState) {
+		super( entityValuedModelPart.getEntityMappingType(), navigablePath );
+		this.resultVariable = resultVariable;
+		this.referencedModelPart = entityValuedModelPart;
+		this.lockMode = creationState.getSqlAstCreationState().determineLockMode( resultVariable );
+
+		final EntityMappingType entityDescriptor = referencedModelPart.getEntityMappingType();
+
+		final TableGroup entityTableGroup = creationState.getSqlAstCreationState().getFromClauseAccess().findTableGroup( navigablePath );
+
+		EntityIdentifierMapping identifierMapping = entityDescriptor.getIdentifierMapping();
+		if ( identifierMapping instanceof SingleAttributeIdentifierMapping ) {
+			identifierMapping.createDomainResult(
+					navigablePath.append( EntityIdentifierMapping.ROLE_LOCAL_NAME ),
+					entityTableGroup,
+					null,
+					creationState
+			);
+		}
+		else {
+			visitCompositeIdentifierMapping( navigablePath, creationState, identifierMapping, entityTableGroup );
+		}
+
+		final EntityDiscriminatorMapping discriminatorMapping = getDiscriminatorMapping( entityDescriptor, entityTableGroup );
+		if ( discriminatorMapping != null ) {
+			discriminatorResult = discriminatorMapping.createDomainResult(
+					navigablePath.append( EntityDiscriminatorMapping.ROLE_NAME ),
+					entityTableGroup,
+					null,
+					creationState
+			);
+		}
+		else {
+			discriminatorResult = null;
+		}
+
+		final EntityVersionMapping versionDescriptor = entityDescriptor.getVersionMapping();
+		if ( versionDescriptor == null ) {
+			versionResult = null;
+		}
+		else {
+			versionResult = versionDescriptor.createDomainResult(
+					navigablePath.append( versionDescriptor.getFetchableName() ),
+					entityTableGroup,
+					null,
+					creationState
+			);
+		}
+
+		afterInitialize( creationState );
+	}
+
+	private void visitCompositeIdentifierMapping(
+			NavigablePath navigablePath,
+			DomainResultCreationState creationState,
+			EntityIdentifierMapping identifierMapping,
+			TableGroup entityTableGroup) {
+		ManagedMappingType mappingType = (ManagedMappingType) identifierMapping.getPartMappingType();
+		fetches = new ArrayList<>();
+		mappingType.visitAttributeMappings(
+				attributeMapping -> {
+					if ( attributeMapping instanceof ToOneAttributeMapping ) {
+						((ToOneAttributeMapping)attributeMapping).getForeignKeyDescriptor().createDomainResult(
+								navigablePath.append( EntityIdentifierMapping.ROLE_LOCAL_NAME ),
+								entityTableGroup,
+								null,
+								creationState
+						);
+					}
+					else {
+						attributeMapping.createDomainResult(
+								navigablePath.append( EntityIdentifierMapping.ROLE_LOCAL_NAME ),
+								entityTableGroup,
+								null,
+								creationState
+						);
+					}
+				}
+		);
+	}
+
+	protected EntityDiscriminatorMapping getDiscriminatorMapping(
+			EntityMappingType entityDescriptor,
+			TableGroup entityTableGroup) {
+		return entityDescriptor.getDiscriminatorMapping();
+	}
+
+	@Override
+	public EntityMappingType getReferencedMappingContainer() {
+		return getEntityValuedModelPart().getEntityMappingType();
+	}
+
+	@Override
+	public EntityValuedModelPart getEntityValuedModelPart() {
+		return referencedModelPart;
+	}
+
+	@Override
+	public JavaTypeDescriptor getResultJavaTypeDescriptor() {
+		return getEntityValuedModelPart().getEntityMappingType().getMappedJavaTypeDescriptor();
+	}
+
+	public LockMode getLockMode() {
+		return lockMode;
+	}
+
+	public DomainResult getDiscriminatorResult() {
+		return discriminatorResult;
+	}
+
+	public DomainResult getVersionResult() {
+		return versionResult;
+	}
+
+	@Override
+	public FetchableContainer getReferencedMappingType() {
+		return getReferencedMappingContainer();
+	}
+
+	@Override
+	public EntityValuedModelPart getReferencedModePart() {
+		return getEntityValuedModelPart();
+	}
+
+	@Override
+	public String getResultVariable() {
+		return resultVariable;
+	}
+
+	@Override
+	public DomainResultAssembler createResultAssembler(AssemblerCreationState creationState) {
+		// todo (6.0) : seems like here is where we ought to determine the SQL selection mappings
+
+		final EntityInitializer initializer = (EntityInitializer) creationState.resolveInitializer(
+				getNavigablePath(),
+				() -> new EntityResultInitializer(
+						this,
+						getNavigablePath(),
+						getLockMode(),
+						identifierResult,
+						getDiscriminatorResult(),
+						getVersionResult(),
+						creationState
+				)
+		);
+
+		return new EntityAssembler( getResultJavaTypeDescriptor(), initializer );
+	}
+
+	@Override
+	public String toString() {
+		return "EntityResultImpl {" + getNavigablePath() + "}";
+	}
+}

hibernate-core/src/main/java/org/hibernate/sql/results/internal/StandardRowReader.java

@@ -138,25 +138,27 @@ public class StandardRowReader<T> implements RowReader<T> {
 		// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 		// old
 
-		for ( int i = 0; i < initializers.size(); i++ ) {
+		final int numberOfInitializers = initializers.size();
+
+		for ( int i = 0; i < numberOfInitializers; i++ ) {
 			final Initializer initializer = initializers.get( i );
 			if ( ! ( initializer instanceof CollectionInitializer ) ) {
 				initializer.resolveKey( rowProcessingState );
 			}
 		}
 
-		for ( int i = 0; i < initializers.size(); i++ ) {
+		for ( int i = 0; i < numberOfInitializers; i++ ) {
 			final Initializer initializer = initializers.get( i );
 			if ( initializer instanceof CollectionInitializer ) {
 				initializer.resolveKey( rowProcessingState );
 			}
 		}
 
-		for ( int i = 0; i < initializers.size(); i++ ) {
+		for ( int i = 0; i < numberOfInitializers; i++ ) {
 			initializers.get( i ).resolveInstance( rowProcessingState );
 		}
 
-		for ( int i = 0; i < initializers.size(); i++ ) {
+		for ( int i = 0; i < numberOfInitializers; i++ ) {
 			initializers.get( i ).initializeInstance( rowProcessingState );
 		}
 	}

hibernate-core/src/main/java/org/hibernate/sql/results/internal/domain/CircularBiDirectionalFetchImpl.java

@@ -39,24 +39,27 @@ import org.hibernate.type.descriptor.java.JavaTypeDescriptor;
 /**
  * @author Andrea Boriero
  */
-public class BiDirectionalFetchImpl implements BiDirectionalFetch, Association {
+public class CircularBiDirectionalFetchImpl implements BiDirectionalFetch, Association {
 	private final FetchTiming timing;
 	private final NavigablePath navigablePath;
 	private final Fetchable fetchable;
 
 	private final FetchParent fetchParent;
+	private final LockMode lockMode;
 	private final NavigablePath referencedNavigablePath;
 
-	public BiDirectionalFetchImpl(
+	public CircularBiDirectionalFetchImpl(
 			FetchTiming timing,
 			NavigablePath navigablePath,
 			FetchParent fetchParent,
 			Fetchable fetchable,
+			LockMode lockMode,
 			NavigablePath referencedNavigablePath) {
 		this.timing = timing;
 		this.fetchParent = fetchParent;
 		this.navigablePath = navigablePath;
 		this.fetchable = fetchable;
+		this.lockMode = lockMode;
 		this.referencedNavigablePath = referencedNavigablePath;
 	}
 
@@ -174,24 +177,25 @@ public class BiDirectionalFetchImpl implements BiDirectionalFetch, Association {
 
 		@Override
 		public Object assemble(RowProcessingState rowProcessingState, JdbcValuesSourceProcessingOptions options) {
-			final EntityInitializer initializer = resolveCircularInitializer( rowProcessingState );
+			EntityInitializer initializer = resolveCircularInitializer( rowProcessingState );
 			if ( initializer == null ) {
+				final Initializer parentInitializer = rowProcessingState.resolveInitializer( circularPath );
+				if ( circularPath.getParent() != null ) {
+					initializer = (EntityInitializer) rowProcessingState.resolveInitializer( circularPath.getParent() );
+				}
+				else {
+					assert parentInitializer instanceof CollectionInitializer;
+					final CollectionInitializer circ = (CollectionInitializer) parentInitializer;
+					final CollectionKey collectionKey = circ.resolveCollectionKey( rowProcessingState );
+					final EntityKey entityKey = new EntityKey(
+							collectionKey.getKey(),
+							(EntityPersister) ( (AttributeMapping) fetchable ).getMappedTypeDescriptor()
+					);
 
-				final Initializer parentInitializer = rowProcessingState.resolveInitializer(
+					final SharedSessionContractImplementor session = rowProcessingState.getJdbcValuesSourceProcessingState()
-						circularPath.getParent() );
+							.getSession();
-				assert parentInitializer instanceof CollectionInitializer;
+					return session.getPersistenceContext().getEntity( entityKey );
-				final CollectionInitializer circ = (CollectionInitializer) parentInitializer;
+				}
-				final CollectionKey collectionKey = circ.resolveCollectionKey( rowProcessingState );
-				final EntityKey entityKey = new EntityKey(
-						collectionKey.getKey(),
-						(EntityPersister) ( (AttributeMapping) fetchable ).getMappedTypeDescriptor()
-				);
-
-				final SharedSessionContractImplementor session = rowProcessingState.getJdbcValuesSourceProcessingState()
-						.getSession();
-				return session.getPersistenceContext()
-						.getEntity( entityKey );
-
 			}
 			if ( initializer.getInitializedInstance() == null ) {
 				initializer.resolveKey( rowProcessingState );
@@ -203,6 +207,12 @@ public class BiDirectionalFetchImpl implements BiDirectionalFetch, Association {
 
 		private EntityInitializer resolveCircularInitializer(RowProcessingState rowProcessingState) {
 			final Initializer initializer = rowProcessingState.resolveInitializer( circularPath );
+			if ( initializer instanceof EntityInitializer ) {
+				return (EntityInitializer) initializer;
+			}
+			if ( initializer instanceof CollectionInitializer ) {
+				return null;
+			}
 			final ModelPart initializedPart = initializer.getInitializedPart();
 
 			if ( initializedPart instanceof EntityInitializer ) {
@@ -211,7 +221,7 @@ public class BiDirectionalFetchImpl implements BiDirectionalFetch, Association {
 
 			NavigablePath path = circularPath.getParent();
 			Initializer parentInitializer = rowProcessingState.resolveInitializer( path );
-			while ( !( parentInitializer instanceof EntityInitializer) && path.getParent() != null ) {
+			while ( !( parentInitializer instanceof EntityInitializer ) && path.getParent() != null ) {
 				path = path.getParent();
 				parentInitializer = rowProcessingState.resolveInitializer( path );
 

hibernate-core/src/main/java/org/hibernate/sql/results/internal/domain/CircularFetchImpl.java

@@ -0,0 +1,210 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html
+ */
+package org.hibernate.sql.results.internal.domain;
+
+import org.hibernate.LockMode;
+import org.hibernate.engine.FetchTiming;
+import org.hibernate.metamodel.mapping.Association;
+import org.hibernate.metamodel.mapping.AttributeMapping;
+import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
+import org.hibernate.metamodel.mapping.MappingType;
+import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
+import org.hibernate.metamodel.model.domain.NavigableRole;
+import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.results.graph.AssemblerCreationState;
+import org.hibernate.sql.results.graph.BiDirectionalFetch;
+import org.hibernate.sql.results.graph.DomainResult;
+import org.hibernate.sql.results.graph.DomainResultAssembler;
+import org.hibernate.sql.results.graph.DomainResultCreationState;
+import org.hibernate.sql.results.graph.Fetch;
+import org.hibernate.sql.results.graph.FetchOptions;
+import org.hibernate.sql.results.graph.FetchParent;
+import org.hibernate.sql.results.graph.FetchParentAccess;
+import org.hibernate.sql.results.graph.Fetchable;
+import org.hibernate.sql.results.graph.entity.EntityInitializer;
+import org.hibernate.sql.results.graph.entity.internal.EntityDelayedFetchInitializer;
+import org.hibernate.sql.results.graph.entity.internal.EntitySelectFetchInitializer;
+import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
+import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
+import org.hibernate.type.descriptor.java.JavaTypeDescriptor;
+
+/**
+ * @author Andrea Boriero
+ */
+public class CircularFetchImpl implements BiDirectionalFetch, Association {
+	private DomainResult keyResult;
+	private final EntityMappingType entityMappingType;
+	private final FetchTiming timing;
+	private final NavigablePath navigablePath;
+	private final ToOneAttributeMapping fetchable;
+
+	private final FetchParent fetchParent;
+	private final NavigablePath referencedNavigablePath;
+
+	public CircularFetchImpl(
+			EntityMappingType entityMappingType,
+			FetchTiming timing,
+			NavigablePath navigablePath,
+			FetchParent fetchParent,
+			ToOneAttributeMapping fetchable,
+			NavigablePath referencedNavigablePath,
+			DomainResult keyResult) {
+		this.entityMappingType = entityMappingType;
+		this.timing = timing;
+		this.fetchParent = fetchParent;
+		this.navigablePath = navigablePath;
+		this.referencedNavigablePath = referencedNavigablePath;
+		this.fetchable = fetchable;
+		this.keyResult = keyResult;
+
+	}
+
+	@Override
+	public NavigablePath getNavigablePath() {
+		return navigablePath;
+	}
+
+	@Override
+	public NavigablePath getReferencedPath() {
+		return referencedNavigablePath;
+	}
+
+	@Override
+	public FetchParent getFetchParent() {
+		return fetchParent;
+	}
+
+	@Override
+	public Fetchable getFetchedMapping() {
+		return fetchable;
+	}
+
+	@Override
+	public JavaTypeDescriptor getResultJavaTypeDescriptor() {
+		return fetchable.getJavaTypeDescriptor();
+	}
+
+	@Override
+	public DomainResultAssembler createAssembler(
+			FetchParentAccess parentAccess,
+			AssemblerCreationState creationState) {
+		final DomainResultAssembler resultAssembler = keyResult.createResultAssembler( creationState );
+
+		final EntityInitializer initializer = (EntityInitializer) creationState.resolveInitializer(
+				getNavigablePath(),
+				() -> {
+					if ( timing == FetchTiming.IMMEDIATE ) {
+						return new EntitySelectFetchInitializer(
+								getReferencedPath(),
+								entityMappingType.getEntityPersister(),
+								resultAssembler,
+								fetchable.isNullable()
+						);
+					}
+					else {
+						return new EntityDelayedFetchInitializer(
+								getReferencedPath(),
+								(EntityPersister) ( (AttributeMapping) fetchable ).getMappedTypeDescriptor(),
+								resultAssembler
+						);
+					}
+				}
+		);
+
+		return new BiDirectionalFetchAssembler(
+				initializer,
+				fetchable.getJavaTypeDescriptor()
+		);
+	}
+
+	@Override
+	public FetchTiming getTiming() {
+		return timing;
+	}
+
+	@Override
+	public boolean hasTableGroup() {
+		return true;
+	}
+
+	@Override
+	public String getFetchableName() {
+		return fetchable.getFetchableName();
+	}
+
+	@Override
+	public FetchOptions getMappedFetchOptions() {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public String getPartName() {
+		return fetchable.getFetchableName();
+	}
+
+	@Override
+	public NavigableRole getNavigableRole() {
+		return fetchable.getNavigableRole();
+	}
+
+	@Override
+	public EntityMappingType findContainingEntityMapping() {
+		return fetchable.findContainingEntityMapping();
+	}
+
+	@Override
+	public MappingType getPartMappingType() {
+		return fetchable.getPartMappingType();
+	}
+
+	@Override
+	public JavaTypeDescriptor getJavaTypeDescriptor() {
+		return fetchable.getJavaTypeDescriptor();
+	}
+
+	@Override
+	public ForeignKeyDescriptor getForeignKeyDescriptor() {
+		return ( (Association) fetchParent ).getForeignKeyDescriptor();
+	}
+
+	@Override
+	public Fetch generateFetch(
+			FetchParent fetchParent,
+			NavigablePath fetchablePath,
+			FetchTiming fetchTiming,
+			boolean selected,
+			LockMode lockMode,
+			String resultVariable,
+			DomainResultCreationState creationState) {
+		throw new UnsupportedOperationException();
+	}
+
+	private static class BiDirectionalFetchAssembler implements DomainResultAssembler {
+		private EntityInitializer initializer;
+		private JavaTypeDescriptor assembledJavaTypeDescriptor;
+
+		public BiDirectionalFetchAssembler(
+				EntityInitializer initializer,
+				JavaTypeDescriptor assembledJavaTypeDescriptor) {
+			this.initializer = initializer;
+			this.assembledJavaTypeDescriptor = assembledJavaTypeDescriptor;
+		}
+
+		@Override
+		public Object assemble(RowProcessingState rowProcessingState, JdbcValuesSourceProcessingOptions options) {
+			return initializer.getInitializedInstance();
+		}
+
+		@Override
+		public JavaTypeDescriptor getAssembledJavaTypeDescriptor() {
+			return assembledJavaTypeDescriptor;
+		}
+	}
+
+}

hibernate-core/src/main/java/org/hibernate/sql/results/jdbc/internal/JdbcValuesSourceProcessingStateStandardImpl.java

@@ -18,6 +18,7 @@ import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.event.spi.EventSource;
 import org.hibernate.event.spi.PostLoadEvent;
 import org.hibernate.event.spi.PreLoadEvent;
+import org.hibernate.sql.results.graph.Initializer;
 import org.hibernate.sql.results.graph.collection.internal.ArrayInitializer;
 import org.hibernate.query.spi.QueryOptions;
 import org.hibernate.sql.exec.spi.ExecutionContext;
@@ -41,6 +42,7 @@ public class JdbcValuesSourceProcessingStateStandardImpl implements JdbcValuesSo
 	private final BiConsumer<EntityKey,LoadingEntityEntry> loadingEntityEntryConsumer;
 
 	private Map<EntityKey, LoadingEntityEntry> loadingEntityMap;
+	private Map<EntityKey, Initializer> initializerMap;
 	private Map<CollectionKey, LoadingCollectionEntry> loadingCollectionMap;
 	private List<CollectionInitializer> arrayInitializers;
 
@@ -105,6 +107,22 @@ public class JdbcValuesSourceProcessingStateStandardImpl implements JdbcValuesSo
 		loadingEntityMap.put( entityKey, loadingEntry );
 	}
 
+	@Override
+	public void registerInitilaizer(
+			EntityKey entityKey,
+			Initializer initializer) {
+		if ( initializerMap == null ) {
+			initializerMap = new HashMap<>();
+		}
+		initializerMap.put( entityKey, initializer );
+
+	}
+
+	@Override
+	public Initializer findInitializer(EntityKey entityKey) {
+		return initializerMap == null ? null : initializerMap.get( entityKey );
+	}
+
 	@Override
 	public LoadingEntityEntry findLoadingEntityLocally(EntityKey entityKey) {
 		return loadingEntityMap == null ? null : loadingEntityMap.get( entityKey );

hibernate-core/src/main/java/org/hibernate/sql/results/jdbc/spi/JdbcValuesSourceProcessingState.java

@@ -11,6 +11,7 @@ import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.event.spi.PostLoadEvent;
 import org.hibernate.event.spi.PreLoadEvent;
+import org.hibernate.sql.results.graph.Initializer;
 import org.hibernate.sql.results.spi.LoadContexts;
 import org.hibernate.sql.results.graph.collection.LoadingCollectionEntry;
 import org.hibernate.sql.results.graph.entity.LoadingEntityEntry;
@@ -55,6 +56,13 @@ public interface JdbcValuesSourceProcessingState {
 			EntityKey entityKey,
 			LoadingEntityEntry loadingEntry);
 
+	void registerInitilaizer(
+			EntityKey entityKey,
+			Initializer initializer);
+
+	Initializer findInitializer(EntityKey entityKey);
+
+
 	/**
 	 * Find a LoadingCollectionEntry locally to this context.
 	 *