HHH-18476 Reset resolved entities after resolveEntityKeyOnly and improve loading further

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

@@ -8,9 +8,9 @@ package org.hibernate.metamodel.mapping.internal;
 
 import java.io.Serializable;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
-import java.util.Set;
 import java.util.function.Consumer;
 
 import org.hibernate.MappingException;
@@ -93,6 +93,11 @@ public abstract class AbstractEmbeddableMapping implements EmbeddableMappingType
 		return 0;
 	}
 
+	@Override
+	public Collection<ConcreteEmbeddableType> getConcreteEmbeddableTypes() {
+		return Collections.singleton( this );
+	}
+
 	@Override
 	public boolean declaresAttribute(AttributeMapping attributeMapping) {
 		return true;

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping.internal;
 import java.io.Serializable;
 import java.util.BitSet;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
@@ -871,7 +872,9 @@ public class EmbeddableMappingTypeImpl extends AbstractEmbeddableMapping impleme
 	@Override
 	public Collection<ConcreteEmbeddableType> getConcreteEmbeddableTypes() {
 		//noinspection unchecked
-		return (Collection<ConcreteEmbeddableType>) (Collection<?>) concreteEmbeddableBySubclass.values();
+		return concreteEmbeddableBySubclass == null
+				? Collections.singleton( this )
+				: (Collection<ConcreteEmbeddableType>) (Collection<?>) concreteEmbeddableBySubclass.values();
 	}
 
 	@Override

hibernate-core/src/main/java/org/hibernate/sql/results/graph/collection/internal/AbstractImmediateCollectionInitializer.java

@@ -464,6 +464,19 @@ public abstract class AbstractImmediateCollectionInitializer<Data extends Abstra
 		initializeSubInstancesFromParent( data );
 	}
 
+	@Override
+	public boolean hasLazySubInitializers() {
+		final DomainResultAssembler<?> indexAssembler = getIndexAssembler();
+		if ( indexAssembler != null ) {
+			final Initializer<?> initializer = indexAssembler.getInitializer();
+			if ( initializer != null && ( !initializer.isEager() || initializer.hasLazySubInitializers() ) ) {
+				return true;
+			}
+		}
+		final Initializer<?> initializer = getElementAssembler().getInitializer();
+		return initializer != null && ( !initializer.isEager() || initializer.hasLazySubInitializers() );
+	}
+
 	@Override
 	public void accept(Data data, List<Object> objects) {
 		readCollectionRow( data, objects );

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

@@ -6,9 +6,13 @@
  */
 package org.hibernate.sql.results.graph.embeddable;
 
+import java.util.function.Consumer;
+
 import org.hibernate.metamodel.mapping.EmbeddableValuedModelPart;
 import org.hibernate.sql.results.graph.InitializerData;
 import org.hibernate.sql.results.graph.InitializerParent;
+import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingState;
+import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -34,4 +38,13 @@ public interface EmbeddableInitializer<Data extends InitializerData> extends Ini
 		return this;
 	}
 
+	/**
+	 * Resets the resolved entity registrations by i.e. removing {@link org.hibernate.engine.spi.EntityHolder}.
+	 * 
+	 * This is used after {@link org.hibernate.sql.results.graph.entity.EntityInitializer#resolveEntityKeyOnly(RowProcessingState)}
+	 * to deregister registrations for entities that were only resolved, but not initialized.
+	 * Failing to do this will lead to errors, because {@link org.hibernate.engine.spi.PersistenceContext#postLoad(JdbcValuesSourceProcessingState, Consumer)}
+	 * is called, which expects all registrations to be fully initialized.
+	 */
+	void resetResolvedEntityRegistrations(RowProcessingState rowProcessingState);
 }

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

@@ -27,6 +27,7 @@ import org.hibernate.spi.NavigablePath;
 import org.hibernate.sql.results.graph.AssemblerCreationState;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 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.Initializer;
 import org.hibernate.sql.results.graph.InitializerData;
@@ -64,7 +65,7 @@ public class EmbeddableInitializerImpl extends AbstractInitializer<EmbeddableIni
 	protected final BasicResultAssembler<?> discriminatorAssembler;
 	protected final @Nullable Initializer<InitializerData>[][] subInitializers;
 	protected final @Nullable Initializer<InitializerData>[][] subInitializersForResolveFromInitialized;
-//	protected final BitSet eagerSubInitializers;
+	protected final @Nullable Initializer<InitializerData>[][] collectionContainingSubInitializers;
 	protected final boolean lazyCapable;
 	protected final boolean hasLazySubInitializer;
 
@@ -118,43 +119,83 @@ public class EmbeddableInitializerImpl extends AbstractInitializer<EmbeddableIni
 		// We never want to create empty composites for the FK target or PK, otherwise collections would break
 		this.createEmptyCompositesEnabled = !isPartOfKey && embeddableMappingType.isCreateEmptyCompositesEnabled();
 		this.sessionFactory = creationState.getSqlAstCreationContext().getSessionFactory();
-		this.assemblers = createAssemblers(
+		final Collection<EmbeddableMappingType.ConcreteEmbeddableType> concreteEmbeddableTypes = embeddableMappingType.getConcreteEmbeddableTypes();
-				resultDescriptor,
+		final DomainResultAssembler<?>[][] assemblers = new DomainResultAssembler[concreteEmbeddableTypes.isEmpty() ? 1 : concreteEmbeddableTypes.size()][];
-				creationState,
+		final @Nullable Initializer<InitializerData>[][] subInitializers = new Initializer[assemblers.length][];
-				embeddableMappingType
-		);
-		this.discriminatorAssembler = discriminatorFetch != null ?
-				(BasicResultAssembler<?>) discriminatorFetch.createAssembler( this, creationState ) :
-				null;
-		this.subInitializers = createInitializers( assemblers );
 		final @Nullable Initializer<InitializerData>[][] eagerSubInitializers = new Initializer[subInitializers.length][];
+		final @Nullable Initializer<InitializerData>[][] collectionContainingSubInitializers = new Initializer[subInitializers.length][];
+		final int numberOfFetchables = embeddableMappingType.getNumberOfFetchables();
+
+		Arrays.fill( subInitializers, Initializer.EMPTY_ARRAY );
 		Arrays.fill( eagerSubInitializers, Initializer.EMPTY_ARRAY );
-//		final BitSet eagerSubInitializers = new BitSet(subInitializers.length * embeddableMappingType.getNumberOfFetchables());
+		Arrays.fill( collectionContainingSubInitializers, Initializer.EMPTY_ARRAY );
+		for (int i = 0; i < assemblers.length; i++ ) {
+			assemblers[i] = new DomainResultAssembler[numberOfFetchables];
+		}
+
 		boolean lazyCapable = false;
 		boolean hasLazySubInitializers = false;
-		for ( int subclassId = 0; subclassId < subInitializers.length; subclassId++ ) {
+		for ( int stateArrayPosition = 0; stateArrayPosition < numberOfFetchables; stateArrayPosition++ ) {
-			final Initializer<InitializerData>[] subInitializer = subInitializers[subclassId];
+			final Fetchable stateArrayContributor = embeddableMappingType.getFetchable( stateArrayPosition );
-			for ( int i = 0; i < subInitializer.length; i++ ) {
+			final Fetch fetch = resultDescriptor.findFetch( stateArrayContributor );
-				final Initializer<InitializerData> initializer = subInitializer[i];
+
-				if ( initializer != null ) {
+			final DomainResultAssembler<?> stateAssembler = fetch == null
-					if ( initializer.isEager() ) {
+					? new NullValueAssembler<>( stateArrayContributor.getJavaType() )
-						if ( eagerSubInitializers[subclassId] == Initializer.EMPTY_ARRAY ) {
+					: fetch.createAssembler( this, creationState );
-							eagerSubInitializers[subclassId] = new Initializer[subInitializer.length];
+
-						}
+			if ( concreteEmbeddableTypes.isEmpty() ) {
-						eagerSubInitializers[subclassId][i] = initializer;
+				assemblers[0][stateArrayPosition] = stateAssembler;
-						hasLazySubInitializers = hasLazySubInitializers || initializer.hasLazySubInitializers();
 			}
 			else {
-						hasLazySubInitializers = true;
+				for ( EmbeddableMappingType.ConcreteEmbeddableType concreteEmbeddableType : concreteEmbeddableTypes ) {
-					}
+					if ( concreteEmbeddableType.declaresAttribute( stateArrayPosition ) ) {
-					lazyCapable = lazyCapable || initializer.isLazyCapable();
+						assemblers[concreteEmbeddableType.getSubclassId()][stateArrayPosition] = stateAssembler;
 					}
 				}
 			}
 
+			//noinspection unchecked
+			final Initializer<InitializerData> subInitializer = (Initializer<InitializerData>) stateAssembler.getInitializer();
+			if ( subInitializer != null ) {
+				for (int subclassId = 0; subclassId < assemblers.length; subclassId++ ) {
+					if ( subInitializers[subclassId] == Initializer.EMPTY_ARRAY ) {
+						subInitializers[subclassId] = new Initializer[numberOfFetchables];
+					}
+					subInitializers[subclassId][stateArrayPosition] = subInitializer;
+
+					if ( subInitializer.isEager() ) {
+						if ( eagerSubInitializers[subclassId] == Initializer.EMPTY_ARRAY ) {
+							eagerSubInitializers[subclassId] = new Initializer[numberOfFetchables];
+						}
+						eagerSubInitializers[subclassId][stateArrayPosition] = subInitializer;
+						hasLazySubInitializers = hasLazySubInitializers || subInitializer.hasLazySubInitializers();
+
+						assert fetch != null;
+						final FetchParent fetchParent;
+						if ( ( fetchParent = fetch.asFetchParent() ) != null && fetchParent.containsCollectionFetches()
+								|| subInitializer.isCollectionInitializer() ) {
+							if ( collectionContainingSubInitializers[subclassId] == Initializer.EMPTY_ARRAY ) {
+								collectionContainingSubInitializers[subclassId] = new Initializer[numberOfFetchables];
+							}
+							collectionContainingSubInitializers[subclassId][stateArrayPosition] = subInitializer;
+						}
+					}
+					else {
+						hasLazySubInitializers = true;
+					}
+					lazyCapable = lazyCapable || subInitializer.isLazyCapable();
+				}
+			}
+		}
+		this.assemblers = assemblers;
+		this.discriminatorAssembler = discriminatorFetch != null
+				? (BasicResultAssembler<?>) discriminatorFetch.createAssembler( this, creationState )
+				: null;
+		this.subInitializers = subInitializers;
 		this.subInitializersForResolveFromInitialized = isEnhancedForLazyLoading( embeddableMappingType )
 				? subInitializers
 				: eagerSubInitializers;
+		this.collectionContainingSubInitializers = collectionContainingSubInitializers;
 		this.lazyCapable = lazyCapable;
 		this.hasLazySubInitializer = hasLazySubInitializers;
 	}
@@ -165,61 +206,6 @@ public class EmbeddableInitializerImpl extends AbstractInitializer<EmbeddableIni
 		);
 	}
 
-	protected DomainResultAssembler<?>[][] createAssemblers(
-			EmbeddableResultGraphNode resultDescriptor,
-			AssemblerCreationState creationState,
-			EmbeddableMappingType embeddableTypeDescriptor) {
-		final int size = embeddableTypeDescriptor.getNumberOfFetchables();
-		final DomainResultAssembler<?>[] overallAssemblers = new DomainResultAssembler[size];
-		for ( int i = 0; i < size; i++ ) {
-			final Fetchable stateArrayContributor = embeddableTypeDescriptor.getFetchable( i );
-			final Fetch fetch = resultDescriptor.findFetch( stateArrayContributor );
-
-			final DomainResultAssembler<?> stateAssembler = fetch == null
-					? new NullValueAssembler<>( stateArrayContributor.getJavaType() )
-					: fetch.createAssembler( this, creationState );
-
-			overallAssemblers[i] = stateAssembler;
-		}
-		if ( embeddableTypeDescriptor.isPolymorphic() ) {
-			final Collection<EmbeddableMappingType.ConcreteEmbeddableType> concreteEmbeddableTypes = embeddableTypeDescriptor.getConcreteEmbeddableTypes();
-			final DomainResultAssembler<?>[][] assemblers = new DomainResultAssembler[concreteEmbeddableTypes.size()][];
-			for ( EmbeddableMappingType.ConcreteEmbeddableType concreteEmbeddableType : concreteEmbeddableTypes ) {
-				final DomainResultAssembler<?>[] subAssemblers = new DomainResultAssembler[overallAssemblers.length];
-				for ( int i = 0; i < overallAssemblers.length; i++ ) {
-					if ( concreteEmbeddableType.declaresAttribute( i ) ) {
-						subAssemblers[i] = overallAssemblers[i];
-					}
-				}
-				assemblers[concreteEmbeddableType.getSubclassId()] = subAssemblers;
-			}
-			return assemblers;
-		}
-		return new DomainResultAssembler[][] { overallAssemblers };
-	}
-
-	protected static @Nullable Initializer<InitializerData>[][] createInitializers(DomainResultAssembler<?>[][] assemblers) {
-		@Nullable Initializer<?>[][] subInitializers = new Initializer<?>[assemblers.length][];
-		for ( int i = 0; i < assemblers.length; i++ ) {
-			final DomainResultAssembler<?>[] subAssemblers = assemblers[i];
-			final @Nullable Initializer<?>[] initializers = new Initializer[subAssemblers.length];
-			boolean empty = true;
-			for ( int j = 0; j < subAssemblers.length; j++ ) {
-				final DomainResultAssembler<?> assembler = subAssemblers[j];
-				if ( assembler != null ) {
-					final Initializer<?> initializer = assembler.getInitializer();
-					if ( initializer != null ) {
-						initializers[j] = initializer;
-						empty = false;
-					}
-				}
-			}
-			subInitializers[i] = empty ? Initializer.EMPTY_ARRAY : initializers;
-		}
-		//noinspection unchecked
-		return (Initializer<InitializerData>[][]) subInitializers;
-	}
-
 	@Override
 	public EmbeddableValuedModelPart getInitializedPart() {
 		return embedded;
@@ -271,7 +257,6 @@ public class EmbeddableInitializerImpl extends AbstractInitializer<EmbeddableIni
 		if ( data.getState() != State.UNINITIALIZED ) {
 			return;
 		}
-		// We need to possibly wrap the processing state if the embeddable is within an aggregate
 		data.setInstance( null );
 		if ( discriminatorAssembler != null ) {
 			assert embeddableMappingType.getDiscriminatorMapping() != null;
@@ -296,6 +281,23 @@ public class EmbeddableInitializerImpl extends AbstractInitializer<EmbeddableIni
 		}
 	}
 
+	@Override
+	public void resetResolvedEntityRegistrations(RowProcessingState rowProcessingState) {
+		final EmbeddableInitializerData data = getData( rowProcessingState );
+		for ( Initializer<InitializerData> initializer : subInitializers[data.getSubclassId()] ) {
+			if ( initializer != null ) {
+				final EntityInitializer<?> entityInitializer = initializer.asEntityInitializer();
+				final EmbeddableInitializer<?> embeddableInitializer;
+				if ( entityInitializer != null ) {
+					entityInitializer.resetResolvedEntityRegistrations( rowProcessingState );
+				}
+				else if ( ( embeddableInitializer = initializer.asEmbeddableInitializer() ) != null ) {
+					embeddableInitializer.resetResolvedEntityRegistrations( rowProcessingState );
+				}
+			}
+		}
+	}
+
 	private void resolveKeySubInitializers(EmbeddableInitializerData data) {
 		final RowProcessingState rowProcessingState = data.getRowProcessingState();
 		for ( Initializer<InitializerData> initializer : subInitializers[data.getSubclassId()] ) {
@@ -318,7 +320,8 @@ public class EmbeddableInitializerImpl extends AbstractInitializer<EmbeddableIni
 			}
 			else {
 				final RowProcessingState rowProcessingState = data.getRowProcessingState();
-				for ( Initializer<InitializerData> initializer : subInitializers[data.getSubclassId()] ) {
+				// When a previous row initialized this entity already, we only need to process collections
+				for ( Initializer<InitializerData> initializer : collectionContainingSubInitializers[data.getSubclassId()] ) {
 					if ( initializer != null ) {
 						initializer.resolveFromPreviousRow( rowProcessingState );
 					}

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

@@ -26,12 +26,14 @@ import org.hibernate.spi.NavigablePath;
 import org.hibernate.sql.results.graph.AssemblerCreationState;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 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.Initializer;
 import org.hibernate.sql.results.graph.InitializerData;
 import org.hibernate.sql.results.graph.InitializerParent;
 import org.hibernate.sql.results.graph.embeddable.EmbeddableInitializer;
 import org.hibernate.sql.results.graph.embeddable.EmbeddableResultGraphNode;
+import org.hibernate.sql.results.graph.entity.EntityInitializer;
 import org.hibernate.sql.results.graph.internal.AbstractInitializer;
 import org.hibernate.sql.results.internal.NullValueAssembler;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
@@ -58,6 +60,7 @@ public class NonAggregatedIdentifierMappingInitializer extends AbstractInitializ
 	private final DomainResultAssembler<?>[] assemblers;
 	private final @Nullable Initializer<InitializerData>[] initializers;
 	private final @Nullable Initializer<InitializerData>[] subInitializersForResolveFromInitialized;
+	private final @Nullable Initializer<InitializerData>[] collectionContainingSubInitializers;
 	private final boolean lazyCapable;
 	private final boolean hasLazySubInitializer;
 	private final boolean hasIdClass;
@@ -124,20 +127,41 @@ public class NonAggregatedIdentifierMappingInitializer extends AbstractInitializ
 		this.hasIdClass = embedded.hasContainingClass() && virtualIdEmbeddable != representationEmbeddable;
 
 		this.sessionFactory = creationState.getSqlAstCreationContext().getSessionFactory();
-		this.assemblers = createAssemblers( this, resultDescriptor, creationState, virtualIdEmbeddable, fetchConverter );
+
+		final int size = virtualIdEmbeddable.getNumberOfFetchables();
+		final DomainResultAssembler<?>[] assemblers = new DomainResultAssembler[size];
+		this.assemblers = assemblers;
 		final Initializer<?>[] initializers = new Initializer[assemblers.length];
 		final Initializer<?>[] eagerSubInitializers = new Initializer[assemblers.length];
+		final Initializer<?>[] collectionContainingSubInitializers = new Initializer[assemblers.length];
 		boolean empty = true;
 		boolean emptyEager = true;
+		boolean emptyCollectionInitializers = true;
 		boolean lazyCapable = false;
 		boolean hasLazySubInitializers = false;
-		for ( int i = 0; i < assemblers.length; i++ ) {
+		for ( int i = 0; i < size; i++ ) {
-			final Initializer<?> initializer = assemblers[i].getInitializer();
+			final Fetchable stateArrayContributor = virtualIdEmbeddable.getFetchable( i );
+			final Fetch fetch = fetchConverter.apply( resultDescriptor.findFetch( stateArrayContributor ) );
+
+			final DomainResultAssembler<?> stateAssembler = fetch == null
+					? new NullValueAssembler<>( stateArrayContributor.getJavaType() )
+					: fetch.createAssembler( this, creationState );
+
+			assemblers[i] = stateAssembler;
+
+			final Initializer<?> initializer = stateAssembler.getInitializer();
 			if ( initializer != null ) {
 				if ( initializer.isEager() ) {
 					eagerSubInitializers[i] = initializer;
 					hasLazySubInitializers = hasLazySubInitializers || initializer.hasLazySubInitializers();
 					emptyEager = false;
+					assert fetch != null;
+					final FetchParent fetchParent;
+					if ( ( fetchParent = fetch.asFetchParent() ) != null && fetchParent.containsCollectionFetches()
+							|| initializer.isCollectionInitializer() ) {
+						collectionContainingSubInitializers[i] = initializer;
+						emptyCollectionInitializers = false;
+					}
 				}
 				else {
 					hasLazySubInitializers = true;
@@ -152,34 +176,18 @@ public class NonAggregatedIdentifierMappingInitializer extends AbstractInitializ
 				empty ? Initializer.EMPTY_ARRAY : initializers
 		);
 		// No need to think about bytecode enhancement here, since ids can't contain lazy basic attributes
+		//noinspection unchecked
 		this.subInitializersForResolveFromInitialized = (Initializer<InitializerData>[]) (
 				emptyEager ? Initializer.EMPTY_ARRAY : initializers
 		);
+		//noinspection unchecked
+		this.collectionContainingSubInitializers = (Initializer<InitializerData>[]) (
+				emptyCollectionInitializers ? Initializer.EMPTY_ARRAY : collectionContainingSubInitializers
+		);
 		this.lazyCapable = lazyCapable;
 		this.hasLazySubInitializer = hasLazySubInitializers;
 	}
 
-	protected static DomainResultAssembler<?>[] createAssemblers(
-			InitializerParent<?> parent,
-			EmbeddableResultGraphNode resultDescriptor,
-			AssemblerCreationState creationState,
-			EmbeddableMappingType embeddableTypeDescriptor,
-			Function<Fetch, Fetch> fetchConverter) {
-		final int size = embeddableTypeDescriptor.getNumberOfFetchables();
-		final DomainResultAssembler<?>[] assemblers = new DomainResultAssembler[size];
-		for ( int i = 0; i < size; i++ ) {
-			final Fetchable stateArrayContributor = embeddableTypeDescriptor.getFetchable( i );
-			final Fetch fetch = fetchConverter.apply( resultDescriptor.findFetch( stateArrayContributor ) );
-
-			final DomainResultAssembler<?> stateAssembler = fetch == null
-					? new NullValueAssembler<>( stateArrayContributor.getJavaType() )
-					: fetch.createAssembler( parent, creationState );
-
-			assemblers[i] = stateAssembler;
-		}
-		return assemblers;
-	}
-
 	@Override
 	public EmbeddableValuedModelPart getInitializedPart() {
 		return embedded;
@@ -210,7 +218,6 @@ public class NonAggregatedIdentifierMappingInitializer extends AbstractInitializ
 		if ( data.getState() != State.UNINITIALIZED ) {
 			return;
 		}
-		// We need to possibly wrap the processing state if the embeddable is within an aggregate
 		data.setInstance( null );
 		data.setState( State.KEY_RESOLVED );
 		if ( initializers.length == 0 ) {
@@ -232,6 +239,42 @@ public class NonAggregatedIdentifierMappingInitializer extends AbstractInitializ
 		}
 	}
 
+	@Override
+	public void resetResolvedEntityRegistrations(RowProcessingState rowProcessingState) {
+		final NonAggregatedIdentifierMappingInitializerData data = getData( rowProcessingState );
+		for ( Initializer<InitializerData> initializer : initializers ) {
+			if ( initializer != null ) {
+				final EntityInitializer<?> entityInitializer = initializer.asEntityInitializer();
+				final EmbeddableInitializer<?> embeddableInitializer;
+				if ( entityInitializer != null ) {
+					entityInitializer.resetResolvedEntityRegistrations( rowProcessingState );
+				}
+				else if ( ( embeddableInitializer = initializer.asEmbeddableInitializer() ) != null ) {
+					embeddableInitializer.resetResolvedEntityRegistrations( rowProcessingState );
+				}
+			}
+		}
+	}
+
+	@Override
+	public void resolveFromPreviousRow(NonAggregatedIdentifierMappingInitializerData data) {
+		if ( data.getState() == State.UNINITIALIZED ) {
+			if ( data.getInstance() == null ) {
+				data.setState( State.MISSING );
+			}
+			else {
+				final RowProcessingState rowProcessingState = data.getRowProcessingState();
+				// When a previous row initialized this entity already, we only need to process collections
+				for ( Initializer<InitializerData> initializer : collectionContainingSubInitializers ) {
+					if ( initializer != null ) {
+						initializer.resolveFromPreviousRow( rowProcessingState );
+					}
+				}
+				data.setState( State.INITIALIZED );
+			}
+		}
+	}
+
 	@Override
 	public void resolveInstance(NonAggregatedIdentifierMappingInitializerData data) {
 		if ( data.getState() != State.KEY_RESOLVED ) {
@@ -253,6 +296,9 @@ public class NonAggregatedIdentifierMappingInitializer extends AbstractInitializ
 		else {
 			data.setInstance( embeddableInstantiator.instantiate( data, sessionFactory ) );
 		}
+		if ( parent == null ) {
+			data.setState( State.INITIALIZED );
+		}
 	}
 
 	@Override

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

@@ -62,6 +62,14 @@ public interface EntityInitializer<Data extends InitializerData> extends Initial
 		return getEntityIdentifier( getData( rowProcessingState ) );
 	}
 
+	/**
+	 * Resets the resolved entity registrations by i.e. removing {@link org.hibernate.engine.spi.EntityHolder}.
+	 *
+	 * @see org.hibernate.sql.results.graph.embeddable.EmbeddableInitializer#resetResolvedEntityRegistrations(RowProcessingState)
+	 */
+	default void resetResolvedEntityRegistrations(RowProcessingState rowProcessingState) {
+	}
+
 	@Override
 	default boolean isEntityInitializer() {
 		return true;

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

@@ -63,11 +63,13 @@ 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.Fetch;
+import org.hibernate.sql.results.graph.FetchParent;
 import org.hibernate.sql.results.graph.Initializer;
 import org.hibernate.sql.results.graph.InitializerData;
 import org.hibernate.sql.results.graph.InitializerParent;
 import org.hibernate.sql.results.graph.basic.BasicResultAssembler;
 import org.hibernate.sql.results.graph.collection.internal.AbstractImmediateCollectionInitializer;
+import org.hibernate.sql.results.graph.embeddable.EmbeddableInitializer;
 import org.hibernate.sql.results.graph.entity.EntityInitializer;
 import org.hibernate.sql.results.graph.entity.EntityResultGraphNode;
 import org.hibernate.sql.results.graph.internal.AbstractInitializer;
@@ -130,8 +132,9 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 	private final @Nullable DomainResultAssembler<Object> rowIdAssembler;
 
 	private final DomainResultAssembler<?>[][] assemblers;
-	private final Initializer<?>[][] subInitializers;
+	private final @Nullable Initializer<?>[][] subInitializers;
-	private final Initializer<?>[][] subInitializersForResolveFromInitialized;
+	private final @Nullable Initializer<?>[][] subInitializersForResolveFromInitialized;
+	private final @Nullable Initializer<?>[][] collectionContainingSubInitializers;
 	private final MutabilityPlan<Object>[][] updatableAttributeMutabilityPlans;
 	private final ImmutableBitSet[] lazySets;
 	private final ImmutableBitSet[] maybeLazySets;
@@ -252,9 +255,7 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 		else {
 			identifierAssembler = identifierFetch.createAssembler( this, creationState );
 			final Initializer<?> initializer = identifierAssembler.getInitializer();
-			// For now, assume key many to ones if the identifier has an initializer
+			hasKeyManyToOne = initializer != null && initializer.isLazyCapable();
-			// todo: improve this
-			hasKeyManyToOne = initializer != null;
 		}
 
 		assert entityDescriptor.hasSubclasses() == (discriminatorFetch != null) : "Discriminator should only be fetched if the entity has subclasses";
@@ -281,6 +282,7 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 		final DomainResultAssembler<?>[][] assemblers = new DomainResultAssembler[subMappingTypes.size() + 1][];
 		final Initializer<?>[][] subInitializers = new Initializer<?>[subMappingTypes.size() + 1][];
 		final Initializer<?>[][] eagerSubInitializers = new Initializer<?>[subMappingTypes.size() + 1][];
+		final Initializer<?>[][] collectionContainingSubInitializers = new Initializer<?>[subMappingTypes.size() + 1][];
 		final BitSet[] lazySets = new BitSet[subMappingTypes.size() + 1];
 		final BitSet[] maybeLazySets = new BitSet[subMappingTypes.size() + 1];
 		final MutabilityPlan[][] updatableAttributeMutabilityPlans = new MutabilityPlan[subMappingTypes.size() + 1][];
@@ -310,6 +312,7 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 				if ( subInitializers[subclassId] == null ) {
 					subInitializers[subclassId] = new Initializer<?>[size];
 					eagerSubInitializers[subclassId] = new Initializer<?>[size];
+					collectionContainingSubInitializers[subclassId] = new Initializer<?>[size];
 					lazySets[subclassId] = new BitSet( size );
 					maybeLazySets[subclassId] = new BitSet( size );
 				}
@@ -320,6 +323,12 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 						maybeLazySets[subclassId].set( stateArrayPosition );
 						hasLazySubInitializers = true;
 					}
+					assert fetch != null;
+					final FetchParent fetchParent;
+					if ( ( fetchParent = fetch.asFetchParent() ) != null && fetchParent.containsCollectionFetches()
+							|| subInitializer.isCollectionInitializer() ) {
+						collectionContainingSubInitializers[subclassId][stateArrayPosition] = subInitializer;
+					}
 				}
 				else {
 					// Lazy initializer
@@ -341,52 +350,71 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 					if ( subInitializers[subMappingType.getSubclassId()] == null ) {
 						subInitializers[subMappingType.getSubclassId()] = new Initializer<?>[size];
 						eagerSubInitializers[subMappingType.getSubclassId()] = new Initializer<?>[size];
+						collectionContainingSubInitializers[subMappingType.getSubclassId()] = new Initializer<?>[size];
 						lazySets[subMappingType.getSubclassId()] = new BitSet(size);
 						maybeLazySets[subMappingType.getSubclassId()] = new BitSet(size);
 					}
 					subInitializers[subMappingType.getSubclassId()][stateArrayPosition] = subInitializer;
-					if ( subInitializer.isEager() ) {
+					eagerSubInitializers[subMappingType.getSubclassId()][stateArrayPosition] = eagerSubInitializers[subclassId][stateArrayPosition];
-						eagerSubInitializers[subMappingType.getSubclassId()][stateArrayPosition] = subInitializer;
+					collectionContainingSubInitializers[subMappingType.getSubclassId()][stateArrayPosition] = collectionContainingSubInitializers[subclassId][stateArrayPosition];
-						if ( subInitializer.hasLazySubInitializers() ) {
+					if (lazySets[subclassId].get( stateArrayPosition ) ) {
-							maybeLazySets[subMappingType.getSubclassId()].set( stateArrayPosition );
-						}
-					}
-					else {
 						lazySets[subMappingType.getSubclassId()].set( stateArrayPosition );
+					}
+					if (maybeLazySets[subclassId].get( stateArrayPosition ) ) {
 						maybeLazySets[subMappingType.getSubclassId()].set( stateArrayPosition );
 					}
 				}
 			}
 		}
 		final BitSet emptyBitSet = new BitSet();
-		OUTER: for ( int i = 0; i < subInitializers.length; i++ ) {
+		for ( int i = 0; i < subInitializers.length; i++ ) {
+			boolean emptySubInitializers = true;
 			if ( subInitializers[i] != null ) {
 				for ( Initializer<?> initializer : subInitializers[i] ) {
 					if ( initializer != null ) {
-						continue OUTER;
+						emptySubInitializers = false;
+						break;
 					}
 				}
 			}
+			if ( emptySubInitializers ) {
 				subInitializers[i] = Initializer.EMPTY_ARRAY;
 				lazySets[i] = emptyBitSet;
 				maybeLazySets[i] = emptyBitSet;
 			}
-		OUTER: for ( int i = 0; i < eagerSubInitializers.length; i++ ) {
+
+			boolean emptyContainingSubInitializers = true;
+			if ( collectionContainingSubInitializers[i] != null ) {
+				for ( Initializer<?> initializer : collectionContainingSubInitializers[i] ) {
+					if ( initializer != null ) {
+						emptyContainingSubInitializers = false;
+						break;
+					}
+				}
+			}
+			if ( emptyContainingSubInitializers ) {
+				collectionContainingSubInitializers[i] = Initializer.EMPTY_ARRAY;
+			}
+			boolean emptyEagerSubInitializers = true;
 			if ( eagerSubInitializers[i] != null ) {
 				for ( Initializer<?> initializer : eagerSubInitializers[i] ) {
 					if ( initializer != null ) {
-						continue OUTER;
+						emptyEagerSubInitializers = false;
+						break;
 					}
 				}
 			}
+			if ( emptyEagerSubInitializers ) {
 				eagerSubInitializers[i] = Initializer.EMPTY_ARRAY;
 			}
+		}
 
 		this.assemblers = assemblers;
 		this.subInitializers = subInitializers;
 		this.subInitializersForResolveFromInitialized = rootEntityDescriptor.getBytecodeEnhancementMetadata().isEnhancedForLazyLoading()
 				? subInitializers
 				: eagerSubInitializers;
+		this.collectionContainingSubInitializers = collectionContainingSubInitializers;
 		this.lazySets = Arrays.stream( lazySets ).map( ImmutableBitSet::valueOf ).toArray( ImmutableBitSet[]::new );
 		this.maybeLazySets = Arrays.stream( maybeLazySets )
 				.map( ImmutableBitSet::valueOf )
@@ -431,13 +459,14 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 
 	@Override
 	public @Nullable EntityKey resolveEntityKeyOnly(RowProcessingState rowProcessingState) {
+		assert identifierAssembler != null;
 		final EntityInitializerData data = getData( rowProcessingState );
 		resolveKey( data, true );
+		try {
 			if ( data.getState() == State.MISSING ) {
 				return null;
 			}
 			if ( data.entityKey == null ) {
-			assert identifierAssembler != null;
 				final Object id = identifierAssembler.assemble( rowProcessingState );
 				if ( id == null ) {
 					setMissing( data );
@@ -447,6 +476,32 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 			}
 			return data.entityKey;
 		}
+		finally {
+			final Initializer<?> initializer = identifierAssembler.getInitializer();
+			if ( hasKeyManyToOne && initializer != null ) {
+				final EmbeddableInitializer<?> embeddableInitializer = initializer.asEmbeddableInitializer();
+				assert embeddableInitializer != null;
+				embeddableInitializer.resetResolvedEntityRegistrations( rowProcessingState );
+			}
+		}
+	}
+
+	@Override
+	public void resetResolvedEntityRegistrations(RowProcessingState rowProcessingState) {
+		final EntityInitializerData data = getData( rowProcessingState );
+		if ( data.getState() == State.RESOLVED ) {
+			rowProcessingState.getSession()
+					.getPersistenceContextInternal()
+					.removeEntityHolder( data.entityKey );
+			rowProcessingState.getJdbcValuesSourceProcessingState()
+					.getLoadingEntityHolders()
+					.remove( data.entityHolder );
+			data.entityKey = null;
+			data.entityHolder = null;
+			data.entityInstanceForNotify = null;
+			data.setInstance( null );
+		}
+	}
 
 	protected void resolveKey(EntityInitializerData data, boolean entityKeyOnly) {
 		// todo (6.0) : atm we do not handle sequential selects
@@ -510,16 +565,15 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 
 		if ( oldEntityKey != null && previousRowReuse && oldEntityInstance != null
 				&& areKeysEqual( oldEntityKey.getIdentifier(), id ) ) {
-			// The row we read previously referred to this entity already, so we can safely assume it's initialized.
+			data.setState( State.INITIALIZED );
-			// Unfortunately we can't set the state to INITIALIZED though, as that has other implications,
-			// but RESOLVED is fine, since the EntityEntry is marked as initialized which skips instance initialization
-			data.setState( State.RESOLVED );
 			data.entityKey = oldEntityKey;
 			data.setInstance( oldEntityInstance );
 			data.entityInstanceForNotify = oldEntityInstanceForNotify;
 			data.concreteDescriptor = oldEntityKey.getPersister();
 			data.entityHolder = oldEntityHolder;
+			if ( !entityKeyOnly ) {
 				notifySubInitializersToReusePreviousRowInstance( data );
+			}
 			return;
 		}
 		resolveEntityKey( data, id );
@@ -555,17 +609,23 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 
 	protected void resolveInstanceSubInitializers(EntityInitializerData data) {
 		final int subclassId = data.concreteDescriptor.getSubclassId();
-		final Initializer<?>[] initializers = subInitializersForResolveFromInitialized[subclassId];
+		final EntityEntry entityEntry = data.entityHolder.getEntityEntry();
-		final EntityEntry entityEntry;
+		final Initializer<?>[] initializers;
 		final ImmutableBitSet maybeLazySet;
+		if ( data.entityHolder.getEntityInitializer() == this ) {
+			// When a previous row initialized this entity already, we only need to process collections
+			initializers = collectionContainingSubInitializers[subclassId];
+			maybeLazySet = null;
+		}
+		else {
+			initializers = subInitializersForResolveFromInitialized[subclassId];
+			maybeLazySet = entityEntry.getMaybeLazySet();
 			// Skip resolving if this initializer has no sub-initializers
-		if ( initializers.length == 0
+			// or the lazy set of this initializer is a superset/contains the entity entry maybeLazySet
-				// or the entity entry has a lazy set available
+			if ( initializers.length == 0 || maybeLazySet != null && lazySets[subclassId].contains( maybeLazySet ) ) {
-				|| ( maybeLazySet = ( entityEntry = data.entityHolder.getEntityEntry() ).getMaybeLazySet() ) != null
-				// which is contained in the lazy sub-initializers
-				&& lazySets[subclassId].contains( maybeLazySet ) ) {
 				return;
 			}
+		}
 		final RowProcessingState rowProcessingState = data.getRowProcessingState();
 		assert entityEntry == rowProcessingState.getSession()
 				.getPersistenceContextInternal()
@@ -603,9 +663,24 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 	}
 
 	private void notifySubInitializersToReusePreviousRowInstance(EntityInitializerData data) {
+		final EntityEntry entityEntry = data.entityHolder.getEntityEntry();
+		final Initializer<?>[] subInitializer;
+		final ImmutableBitSet maybeLazySet;
+		if ( data.entityHolder.getEntityInitializer() == this ) {
+			// When a previous row initialized this entity already, we only need to process collections
+			subInitializer = collectionContainingSubInitializers[data.concreteDescriptor.getSubclassId()];
+			maybeLazySet = null;
+		}
+		else {
+			subInitializer = subInitializersForResolveFromInitialized[data.concreteDescriptor.getSubclassId()];
+			maybeLazySet = entityEntry == null ? null : entityEntry.getMaybeLazySet();
+		}
 		final RowProcessingState rowProcessingState = data.getRowProcessingState();
-		for ( Initializer<?> initializer : subInitializers[data.concreteDescriptor.getSubclassId()] ) {
+		for ( int i = 0; i < subInitializer.length; i++ ) {
-			if ( initializer != null ) {
+			final Initializer<?> initializer = subInitializer[i];
+			// It is vital to only resolveFromPreviousRow only for the initializers where the state is maybe lazy,
+			// as the initialization process for the previous row also only called those initializers
+			if ( initializer != null && ( maybeLazySet == null || maybeLazySet.get( i ) ) ) {
 				initializer.resolveFromPreviousRow( rowProcessingState );
 			}
 		}
@@ -674,7 +749,7 @@ public class EntityInitializerImpl extends AbstractInitializer<EntityInitializer
 				setMissing( data );
 			}
 			else {
-				data.setState( State.RESOLVED );
+				data.setState( State.INITIALIZED );
 				notifySubInitializersToReusePreviousRowInstance( data );
 			}
 		}

hibernate-core/src/test/java/org/hibernate/orm/test/query/ReloadWithPreviousRowEntityTest2.java

@@ -0,0 +1,193 @@
+/*
+ * 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.orm.test.query;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.hibernate.testing.orm.junit.EntityManagerFactoryScope;
+import org.hibernate.testing.orm.junit.Jira;
+import org.hibernate.testing.orm.junit.Jpa;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.OneToMany;
+
+@Jpa(
+		annotatedClasses = { ReloadWithPreviousRowEntityTest2.TeacherDetails.class, ReloadWithPreviousRowEntityTest2.Student.class, ReloadWithPreviousRowEntityTest2.Teacher.class }
+)
+@Jira("https://hibernate.atlassian.net/browse/HHH-18271")
+public class ReloadWithPreviousRowEntityTest2 {
+
+	@BeforeEach
+	public void prepareTestData(EntityManagerFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> {
+					Student mathStudent = new Student(16);
+					Student frenchStudent = new Student(17);
+					Student scienceStudent = new Student(18);
+
+					Teacher teacherWithNoStudents = new Teacher(16);
+					Teacher teacherWithOneStudent = new Teacher(17);
+					Teacher teacherWithTwoStudents = new Teacher(18);
+
+					session.persist( teacherWithNoStudents );
+					session.persist( teacherWithOneStudent );
+					session.persist( teacherWithTwoStudents );
+
+					mathStudent.setTeacher( teacherWithOneStudent );
+					teacherWithOneStudent.addStudent( mathStudent );
+
+					frenchStudent.setTeacher( teacherWithTwoStudents );
+					teacherWithTwoStudents.addStudent( frenchStudent );
+
+					scienceStudent.setTeacher( teacherWithTwoStudents );
+					teacherWithTwoStudents.addStudent( scienceStudent );
+
+					session.persist( mathStudent );
+					session.persist( frenchStudent );
+					session.persist( scienceStudent );
+				}
+		);
+	}
+
+	@AfterEach
+	public void dropTestData(EntityManagerFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> {
+					session.createQuery( "delete from Student" ).executeUpdate();
+					session.createQuery( "delete from Teacher" ).executeUpdate();
+					session.createQuery( "delete from TeacherDetails" ).executeUpdate();
+				}
+		);
+	}
+
+	@Test
+	public void testReloadWithPreviousRow(EntityManagerFactoryScope scope) {
+		scope.inEntityManager( em -> {
+			// First load a fully initialized graph i.e. maybeLazySet empty
+			em.createQuery( "select s from Student s join fetch s.teacher t left join fetch t.students left join fetch t.details order by s.id desc", Student.class ).getResultList();
+			// Then load a partially initialized graph and see if previous row optimization works properly
+			em.createQuery( "select s from Student s join fetch s.teacher t left join fetch t.students order by s.id desc", Student.class ).getResultList();
+		} );
+	}
+
+	@Entity(name = "Student")
+	public static class Student {
+		private Integer id;
+
+		private Teacher teacher;
+
+		public Student() {
+		}
+
+		public Student(Integer id) {
+			this.id = id;
+		}
+
+		@Id
+		@Column(name = "student_id")
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+
+		@ManyToOne(optional = false)
+		@JoinColumn(name = "teacher_fk_id")
+		public Teacher getTeacher() {
+			return teacher;
+		}
+
+		public void setTeacher(Teacher teacher) {
+			this.teacher = teacher;
+		}
+
+	}
+
+	@Entity(name = "Teacher")
+	public static class Teacher {
+
+		private Integer id;
+
+		private Set<Student> students = new HashSet<>();
+
+		private TeacherDetails details;
+
+		public Teacher() {
+		}
+
+		public Teacher(Integer id) {
+			this.id = id;
+		}
+
+		@Id
+		@Column(name = "teacher_id")
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+
+		@OneToMany(mappedBy = "teacher")
+		public Set<Student> getStudents() {
+			return students;
+		}
+
+		public void setStudents(Set<Student> students) {
+			this.students = students;
+		}
+
+		public void addStudent(Student student) {
+			students.add( student );
+		}
+
+		@ManyToOne
+		public TeacherDetails getDetails() {
+			return details;
+		}
+
+		public void setDetails(TeacherDetails details) {
+			this.details = details;
+		}
+	}
+
+	@Entity(name = "TeacherDetails")
+	public static class TeacherDetails {
+
+		private Integer id;
+		private String name;
+
+		@Id
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
+}

hibernate-core/src/test/java/org/hibernate/orm/test/query/ReloadWithPreviousRowEntityTest3.java

@@ -0,0 +1,198 @@
+/*
+ * 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.orm.test.query;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.hibernate.testing.orm.junit.EntityManagerFactoryScope;
+import org.hibernate.testing.orm.junit.Jira;
+import org.hibernate.testing.orm.junit.Jpa;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.ManyToMany;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.OneToMany;
+
+@Jpa(
+		annotatedClasses = { ReloadWithPreviousRowEntityTest3.Skill.class, ReloadWithPreviousRowEntityTest3.Student.class, ReloadWithPreviousRowEntityTest3.Teacher.class }
+)
+@Jira("https://hibernate.atlassian.net/browse/HHH-18271")
+public class ReloadWithPreviousRowEntityTest3 {
+
+	@BeforeEach
+	public void prepareTestData(EntityManagerFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> {
+					Student mathStudent = new Student(16);
+					Student frenchStudent = new Student(17);
+					Student scienceStudent = new Student(18);
+
+					Teacher teacherWithNoStudents = new Teacher(16);
+					Teacher teacherWithOneStudent = new Teacher(17);
+					Teacher teacherWithTwoStudents = new Teacher(18);
+
+					session.persist( teacherWithNoStudents );
+					session.persist( teacherWithOneStudent );
+					session.persist( teacherWithTwoStudents );
+
+					mathStudent.setTeacher( teacherWithOneStudent );
+					teacherWithOneStudent.addStudent( mathStudent );
+
+					frenchStudent.setTeacher( teacherWithTwoStudents );
+					teacherWithTwoStudents.addStudent( frenchStudent );
+
+					scienceStudent.setTeacher( teacherWithTwoStudents );
+					teacherWithTwoStudents.addStudent( scienceStudent );
+
+					session.persist( mathStudent );
+					session.persist( frenchStudent );
+					session.persist( scienceStudent );
+				}
+		);
+	}
+
+	@AfterEach
+	public void dropTestData(EntityManagerFactoryScope scope) {
+		scope.inTransaction(
+				(session) -> {
+					session.createQuery( "delete from Student" ).executeUpdate();
+					session.createQuery( "delete from Teacher" ).executeUpdate();
+					session.createQuery( "delete from Skill" ).executeUpdate();
+				}
+		);
+	}
+
+	@Test
+	public void testReloadWithPreviousRow(EntityManagerFactoryScope scope) {
+		scope.inEntityManager( em -> {
+			// First load a fully initialized graph i.e. maybeLazySet empty
+			em.createQuery( "select s from Student s join fetch s.teacher t left join fetch t.students left join fetch t.skills order by s.id desc", Student.class ).getResultList();
+			// Then load a partially initialized graph and see if previous row optimization works properly
+			em.createQuery( "select s from Student s join fetch s.teacher t left join fetch t.students order by s.id desc", Student.class ).getResultList();
+		} );
+	}
+
+	@Entity(name = "Student")
+	public static class Student {
+		private Integer id;
+
+		private Teacher teacher;
+
+		public Student() {
+		}
+
+		public Student(Integer id) {
+			this.id = id;
+		}
+
+		@Id
+		@Column(name = "student_id")
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+
+		@ManyToOne(optional = false)
+		@JoinColumn(name = "teacher_fk_id")
+		public Teacher getTeacher() {
+			return teacher;
+		}
+
+		public void setTeacher(Teacher teacher) {
+			this.teacher = teacher;
+		}
+
+	}
+
+	@Entity(name = "Teacher")
+	public static class Teacher {
+
+		private Integer id;
+
+		private Set<Student> students = new HashSet<>();
+
+		private Set<Skill> skills = new HashSet<>();
+
+		public Teacher() {
+		}
+
+		public Teacher(Integer id) {
+			this.id = id;
+		}
+
+		@Id
+		@Column(name = "teacher_id")
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+
+		@OneToMany(mappedBy = "teacher")
+		public Set<Student> getStudents() {
+			return students;
+		}
+
+		public void setStudents(Set<Student> students) {
+			this.students = students;
+		}
+
+		public void addStudent(Student student) {
+			students.add( student );
+		}
+
+		@ManyToMany
+		public Set<Skill> getSkills() {
+			return skills;
+		}
+
+		public void addSkill(Skill skill) {
+			skills.add( skill );
+		}
+
+		public void setSkills(Set<Skill> skills) {
+			this.skills = skills;
+		}
+	}
+
+	@Entity(name = "Skill")
+	public static class Skill {
+
+		private Integer id;
+		private String name;
+
+		@Id
+		public Integer getId() {
+			return id;
+		}
+
+		public void setId(Integer id) {
+			this.id = id;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
+}

hibernate-core/src/test/java/org/hibernate/orm/test/stream/KeyToOneCollectionFetchScrollTest.java

@@ -0,0 +1,172 @@
+/*
+ * 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.orm.test.stream;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.hibernate.testing.orm.junit.DomainModel;
+import org.hibernate.testing.orm.junit.Jira;
+import org.hibernate.testing.orm.junit.SessionFactory;
+import org.hibernate.testing.orm.junit.SessionFactoryScope;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import jakarta.persistence.CascadeType;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.MappedSuperclass;
+import jakarta.persistence.OneToMany;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@DomainModel( annotatedClasses = {
+		KeyToOneCollectionFetchScrollTest.BasicEntity.class,
+		KeyToOneCollectionFetchScrollTest.EntityA.class,
+		KeyToOneCollectionFetchScrollTest.EntityB.class,
+		KeyToOneCollectionFetchScrollTest.EntityC.class,
+} )
+@SessionFactory
+@Jira("https://hibernate.atlassian.net/browse/HHH-18476")
+public class KeyToOneCollectionFetchScrollTest {
+	@BeforeAll
+	public void setUp(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			final EntityA a1 = new EntityA("a1");
+			final EntityA a2 = new EntityA("a2");
+			final EntityB b1 = new EntityB("b1");
+			b1.a1 = a1;
+			b1.a2 = a2;
+			session.persist( a1 );
+			session.persist( a2 );
+			session.persist( b1 );
+			final EntityA a3 = new EntityA("a3");
+			final EntityA a4 = new EntityA("a4");
+			final EntityB b2 = new EntityB("b2");
+			b2.a1 = a3;
+			b2.a2 = a4;
+			session.persist( a3 );
+			session.persist( a4 );
+			session.persist( b2 );
+		} );
+	}
+
+	@AfterAll
+	public void tearDown(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			session.createMutationQuery( "delete from EntityC" ).executeUpdate();
+			session.createMutationQuery( "delete from EntityB" ).executeUpdate();
+			session.createMutationQuery( "delete from EntityA" ).executeUpdate();
+		} );
+	}
+
+	@Test
+	public void testScrollWithKeyToOne(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			try (final Stream<EntityB> stream = session.createQuery(
+					"select b from EntityB b join fetch b.a1 join fetch b.a2 left join fetch b.c c order by b.name",
+					EntityB.class
+			).getResultStream()) {
+				final List<EntityB> list = stream.collect( Collectors.toList() );
+				assertThat( list ).hasSize( 2 );
+				assertThat( list.get( 0 ).getA1() ).isNotNull();
+				assertThat( list.get( 0 ).getC() ).hasSize( 0 );
+			}
+		} );
+	}
+
+	@MappedSuperclass
+	public static abstract class BasicEntity {
+		@Id
+		@GeneratedValue
+		Long id;
+		String name;
+
+		public BasicEntity() {
+		}
+
+		public BasicEntity(String name) {
+			this.name = name;
+		}
+
+		public Long getId() {
+			return id;
+		}
+
+		public String getName() {
+			return name;
+		}
+	}
+
+	@Entity( name = "EntityA" )
+	public static class EntityA extends BasicEntity {
+
+		public EntityA() {
+		}
+
+		public EntityA(String name) {
+			super( name );
+		}
+	}
+
+	@Entity( name = "EntityB" )
+	public static class EntityB {
+		@Id
+		@ManyToOne
+		@JoinColumn( name = "a1_id")
+		EntityA a1;
+		@Id
+		@ManyToOne
+		@JoinColumn( name = "a2_id")
+		EntityA a2;
+		String name;
+		@OneToMany( cascade = CascadeType.PERSIST )
+		@JoinColumn( name = "entityb_a1_id", referencedColumnName = "a1_id")
+		@JoinColumn( name = "entityb_a2_id", referencedColumnName = "a2_id")
+		Set<EntityC> c = new HashSet<>();
+
+		public EntityB() {
+		}
+
+		public EntityB(String name) {
+			this.name = name;
+		}
+
+		public EntityA getA1() {
+			return a1;
+		}
+
+		public EntityA getA2() {
+			return a2;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+		public Set<EntityC> getC() {
+			return c;
+		}
+	}
+
+	@Entity( name = "EntityC" )
+	public static class EntityC extends BasicEntity {
+		public EntityC() {
+		}
+
+		public EntityC(String name) {
+			super( name );
+		}
+	}
+}