rework collection loading state management - remove the "temp" load collections in various PersistentCollection impls

hibernate-core/src/main/java/org/hibernate/collection/internal/PersistentArrayHolder.java

@@ -12,14 +12,14 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
 
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.CoreMessageLogger;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
-import org.hibernate.sql.results.graph.DomainResultAssembler;
-import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 import org.hibernate.type.Type;
 
 import org.jboss.logging.Logger;
@@ -41,7 +41,6 @@ public class PersistentArrayHolder extends AbstractPersistentCollection {
 
 	//just to help out during the load (ugly, i know)
 	private transient Class elementClass;
-	private transient java.util.List tempList;
 
 	/**
 	 * Constructs a PersistentCollection instance for holding an array.
@@ -130,6 +129,15 @@ public class PersistentArrayHolder extends AbstractPersistentCollection {
 		return result;
 	}
 
+	@Override
+	public void injectLoadedState(PluralAttributeMapping attributeMapping, List loadingState) {
+		assert isInitializing();
+		array = Array.newInstance( elementClass, loadingState.size() );
+		for ( int i = 0; i < loadingState.size(); i++ ) {
+			Array.set( array, i, loadingState.get( i ) );
+		}
+	}
+
 	@SuppressWarnings("UnusedDeclaration")
 	public Object getArray() {
 		return array;
@@ -177,32 +185,6 @@ public class PersistentArrayHolder extends AbstractPersistentCollection {
 	}
 
 	@Override
-	public Object readFrom(
-			RowProcessingState rowProcessingState,
-			DomainResultAssembler elementAssembler,
-			DomainResultAssembler indexAssembler,
-			DomainResultAssembler identifierAssembler,
-			Object owner) throws HibernateException {
-		assert elementAssembler != null;
-		assert indexAssembler != null;
-		assert identifierAssembler == null;
-
-		final Object element = elementAssembler.assemble( rowProcessingState );
-		final int index = (int) indexAssembler.assemble( rowProcessingState );
-
-		for ( int i = tempList.size(); i<=index; i++) {
-			//noinspection unchecked
-			tempList.add( i, null );
-		}
-
-		//noinspection unchecked
-		tempList.set( index, element );
-
-		return element;
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
 	public Iterator entries(CollectionPersister persister) {
 		return elements();
 	}
@@ -210,17 +192,11 @@ public class PersistentArrayHolder extends AbstractPersistentCollection {
 	@Override
 	public void beginRead() {
 		super.beginRead();
-		tempList = new ArrayList();
 	}
 
 	@Override
 	public boolean endRead() {
 		setInitialized();
-		array = Array.newInstance( elementClass, tempList.size() );
-		for ( int i=0; i<tempList.size(); i++ ) {
-			Array.set( array, i, tempList.get( i ) );
-		}
-		tempList = null;
 		return true;
 	}
 
@@ -320,14 +296,4 @@ public class PersistentArrayHolder extends AbstractPersistentCollection {
 	public boolean entryExists(Object entry, int i) {
 		return entry != null;
 	}
-
-	public void load(int index, Object element) {
-		assert isInitializing();
-
-		for ( int i = tempList.size(); i <= index; ++i ) {
-			tempList.add( i, null );
-		}
-
-		tempList.set( index, element );
-	}
 }

hibernate-core/src/main/java/org/hibernate/collection/internal/PersistentBag.java

@@ -19,6 +19,7 @@ import java.util.Map;
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
@@ -36,7 +37,9 @@ public class PersistentBag extends AbstractPersistentCollection implements List
 
 	protected List bag;
 
-	// The Collection provided to a PersistentBag constructor,
+	/**
+	 * The Collection provided to a PersistentBag constructor
+	 */
 	private Collection providedCollection;
 
 	/**
@@ -121,22 +124,15 @@ public class PersistentBag extends AbstractPersistentCollection implements List
 		return bag.iterator();
 	}
 
-	@Override
+	public void injectLoadedState(PluralAttributeMapping collectionAttributeMapping, List loadingState) {
-	public Object readFrom(
+		assert isInitializing();
-			RowProcessingState rowProcessingState,
+		assert bag != null;
-			DomainResultAssembler elementAssembler,
+		assert bag.isEmpty();
-			DomainResultAssembler indexAssembler,
-			DomainResultAssembler identifierAssembler,
-			Object owner) throws HibernateException {
-		assert indexAssembler == null;
-		assert identifierAssembler == null;
 
-		final Object element = elementAssembler.assemble( rowProcessingState );
+		for ( int i = 0; i < loadingState.size(); i++ ) {
-		if ( element != null ) {
+			//noinspection unchecked,UseBulkOperation
-			//noinspection unchecked
+			bag.add( loadingState.get( i ) );
-			bag.add( element );
 		}
-		return element;
 	}
 
 	@Override
@@ -645,11 +641,6 @@ public class PersistentBag extends AbstractPersistentCollection implements List
 		return super.hashCode();
 	}
 
-	public void load(Object element) {
-		assert isInitializing();
-		bag.add( element );
-	}
-
 	final class Clear implements DelayedOperation {
 		@Override
 		public void operate() {

hibernate-core/src/main/java/org/hibernate/collection/internal/PersistentIdentifierBag.java

@@ -18,6 +18,7 @@ import java.util.Map;
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
@@ -39,7 +40,9 @@ public class PersistentIdentifierBag extends AbstractPersistentCollection implem
 	protected List<Object> values;
 	protected Map<Integer, Object> identifiers;
 
-	// The Collection provided to a PersistentIdentifierBag constructor,
+	/**
+	 * The Collection provided to a PersistentIdentifierBag constructor
+	 */
 	private Collection providedValues;
 
 	/**
@@ -530,4 +533,24 @@ public class PersistentIdentifierBag extends AbstractPersistentCollection implem
 			values.add( element );
 		}
 	}
+
+	public void injectLoadedState(PluralAttributeMapping attributeMapping, List loadingState) {
+		assert isInitializing();
+		assert identifiers != null;
+		assert identifiers.isEmpty();
+		assert values != null;
+		assert values.isEmpty();
+
+		for ( int i = 0; i < loadingState.size(); i++ ) {
+			final Object[] row = (Object[]) loadingState.get( i );
+			final Object identifier = row[0];
+			final Object element = row[1];
+
+			Object old = identifiers.put( values.size(), identifier );
+			if ( old == null ) {
+				//maintain correct duplication if loaded in a cartesian product
+				values.add( element );
+			}
+		}
+	}
 }

hibernate-core/src/main/java/org/hibernate/collection/internal/PersistentList.java

@@ -16,9 +16,8 @@ import java.util.ListIterator;
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
-import org.hibernate.sql.results.graph.DomainResultAssembler;
-import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 import org.hibernate.type.Type;
 
 /**
@@ -394,33 +393,6 @@ public class PersistentList extends AbstractPersistentCollection implements List
 	}
 
 	@Override
-	public Object readFrom(
-			RowProcessingState rowProcessingState,
-			DomainResultAssembler elementAssembler,
-			DomainResultAssembler indexAssembler,
-			DomainResultAssembler identifierAssembler,
-			Object owner) throws HibernateException {
-		assert elementAssembler != null;
-		assert indexAssembler != null;
-		assert identifierAssembler == null;
-
-		final Object element = elementAssembler.assemble( rowProcessingState );
-		final int index = (int) indexAssembler.assemble( rowProcessingState );
-
-		//pad with nulls from the current last element up to the new index
-		for ( int i = list.size(); i<=index; i++) {
-			//noinspection unchecked
-			list.add( i, null );
-		}
-
-		//noinspection unchecked
-		list.set( index, element );
-
-		return element;
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
 	public Iterator entries(CollectionPersister persister) {
 		return list.iterator();
 	}
@@ -521,6 +493,14 @@ public class PersistentList extends AbstractPersistentCollection implements List
 		return entry!=null;
 	}
 
+	public void injectLoadedState(PluralAttributeMapping collectionAttributeMapping, List loadingStateList) {
+		assert isInitializing();
+		assert list != null;
+
+		//noinspection unchecked
+		list.addAll( loadingStateList );
+	}
+
 	final class Clear implements DelayedOperation {
 		@Override
 		public void operate() {

hibernate-core/src/main/java/org/hibernate/collection/internal/PersistentMap.java

@@ -18,9 +18,8 @@ import java.util.Set;
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
-import org.hibernate.sql.results.graph.DomainResultAssembler;
-import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 import org.hibernate.type.Type;
 
 
@@ -274,50 +273,27 @@ public class PersistentMap extends AbstractPersistentCollection implements Map {
 		return map.toString();
 	}
 
-	private transient List<Object[]> loadingEntries;
-
 	@Override
-	public Object readFrom(
-			RowProcessingState rowProcessingState,
-			DomainResultAssembler elementAssembler,
-			DomainResultAssembler indexAssembler,
-			DomainResultAssembler identifierAssembler,
-			Object owner) throws HibernateException {
-		assert elementAssembler != null;
-		assert indexAssembler != null;
-		assert identifierAssembler == null;
-
-		final Object element = elementAssembler.assemble( rowProcessingState );
-
-		if ( element != null ) {
-			final Object index = indexAssembler.assemble( rowProcessingState );
-			if ( loadingEntries == null ) {
-				loadingEntries = new ArrayList<>();
-			}
-			loadingEntries.add( new Object[] { index, element } );
-		}
-
-		return element;
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
 	public boolean endRead() {
-		if ( loadingEntries != null ) {
-			for ( Object[] entry : loadingEntries ) {
-				map.put( entry[0], entry[1] );
-			}
-			loadingEntries = null;
-		}
 		return super.endRead();
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
 	public Iterator entries(CollectionPersister persister) {
 		return map.entrySet().iterator();
 	}
 
+	public void injectLoadedState(PluralAttributeMapping collectionAttributeMapping, List loadingState) {
+		assert isInitializing();
+		assert map != null;
+
+		for ( int i = 0; i < loadingState.size(); i++ ) {
+			final Object[] keyVal = (Object[]) loadingState.get( i );
+			//noinspection unchecked
+			map.put( keyVal[0], keyVal[1] );
+		}
+	}
+
 	/**
 	 * a wrapper for Map.Entry sets
 	 */

hibernate-core/src/main/java/org/hibernate/collection/internal/PersistentSet.java

@@ -17,6 +17,7 @@ import java.util.Set;
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
@@ -32,7 +33,6 @@ import org.hibernate.type.Type;
  */
 public class PersistentSet extends AbstractPersistentCollection implements java.util.Set {
 	protected Set set;
-	protected transient List tempList;
 
 	/**
 	 * Empty constructor.
@@ -141,12 +141,6 @@ public class PersistentSet extends AbstractPersistentCollection implements java.
 		this.set = (Set) persister.getCollectionType().instantiate( anticipatedSize );
 	}
 
-	@SuppressWarnings("unchecked")
-	public void load(Object element) {
-		assert isInitializing();
-		tempList.add( element );
-	}
-
 	@Override
 	@SuppressWarnings("unchecked")
 	public void initializeFromCache(CollectionPersister persister, Object disassembled, Object owner)
@@ -312,7 +306,6 @@ public class PersistentSet extends AbstractPersistentCollection implements java.
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
 	public void clear() {
 		if ( isClearQueueEnabled() ) {
 			queueOperation( new Clear() );
@@ -327,51 +320,25 @@ public class PersistentSet extends AbstractPersistentCollection implements java.
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
 	public String toString() {
 		read();
 		return set.toString();
 	}
 
-	@Override
+	public void injectLoadedState(
-	public Object readFrom(
+			PluralAttributeMapping attributeMapping,
-			RowProcessingState rowProcessingState,
+			List loadingStateList) {
-			DomainResultAssembler elementAssembler,
+		final CollectionPersister collectionDescriptor = attributeMapping.getCollectionDescriptor();
-			DomainResultAssembler indexAssembler,
+		this.set = (Set) attributeMapping.getCollectionDescriptor().getCollectionSemantics().instantiateRaw(
-			DomainResultAssembler identifierAssembler,
+				loadingStateList.size(),
-			Object owner) throws HibernateException {
+				collectionDescriptor
-		assert elementAssembler != null;
+		);
-		assert indexAssembler == null;
-		assert identifierAssembler == null;
 
-		final Object element = elementAssembler.assemble( rowProcessingState );
+		//noinspection unchecked
-
+		this.set.addAll( loadingStateList );
-		if ( element != null ) {
-			//noinspection unchecked
-			tempList.add( element );
-		}
-
-		return element;
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
-	public void beginRead() {
-		super.beginRead();
-		tempList = new ArrayList();
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
-	public boolean endRead() {
-		set.addAll( tempList );
-		tempList = null;
-		// ensure that operationQueue is considered
-		return super.endRead();
-	}
-
-	@Override
-	@SuppressWarnings("unchecked")
 	public Iterator entries(CollectionPersister persister) {
 		return set.iterator();
 	}

hibernate-core/src/main/java/org/hibernate/collection/spi/PersistentCollection.java

@@ -9,9 +9,11 @@ package org.hibernate.collection.spi;
 import java.io.Serializable;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 
 import org.hibernate.HibernateException;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
@@ -91,6 +93,8 @@ public interface PersistentCollection {
 
 	/**
 	 * Called just before reading any rows from the JDBC result set
+	 *
+	 * todo (6.0) remove these.  should no longer be a need with change in how collections are initialized
 	 */
 	void beginRead();
 
@@ -98,6 +102,8 @@ public interface PersistentCollection {
 	 * Called after reading all rows from the JDBC result set
 	 *
 	 * @return Whether to end the read.
+	 *
+	 * todo (6.0) remove these.  should no longer be a need with change in how collections are initialized
 	 */
 	boolean endRead();
 
@@ -105,6 +111,8 @@ public interface PersistentCollection {
 	 * Called after initializing from cache
 	 *
 	 * @return ??
+	 *
+	 * todo (6.0) remove these.  should no longer be a need with change in how collections are initialized
 	 */
 	boolean afterInitialize();
 
@@ -158,13 +166,18 @@ public interface PersistentCollection {
 	 * Read a row from the JDBC values
 	 *
 	 * @throws HibernateException Generally indicates a problem resolving data read from the ResultSet
+	 *
+	 *
+	  todo (6.0) remove these.  should no longer be a need with change in how collections are initialized
 	 */
-	Object readFrom(
+	default Object readFrom(
 			RowProcessingState rowProcessingState,
 			DomainResultAssembler elementAssembler,
 			DomainResultAssembler indexAssembler,
 			DomainResultAssembler identifierAssembler,
-			Object owner) throws HibernateException;
+			Object owner) throws HibernateException {
+		return null;
+	}
 
 	/**
 	 * Get the identifier of the given collection entry.  This refers to the collection identifier, not the
@@ -458,4 +471,8 @@ public interface PersistentCollection {
 	 */
 	Collection getOrphans(Serializable snapshot, String entityName);
 
+	/**
+	 * Inject the state loaded for a collection instance.
+	 */
+	void injectLoadedState(PluralAttributeMapping attributeMapping, List loadingState);
 }

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

@@ -7,6 +7,8 @@
 package org.hibernate.sql.results.graph.collection;
 
 import java.io.Serializable;
+import java.util.List;
+import java.util.function.Consumer;
 
 import org.hibernate.collection.spi.PersistentCollection;
 import org.hibernate.persister.collection.CollectionPersister;
@@ -18,13 +20,33 @@ import org.hibernate.sql.exec.spi.ExecutionContext;
  * @author Steve Ebersole
  */
 public interface LoadingCollectionEntry {
+	/**
+	 * The descriptor for the collection being loaded
+	 */
 	CollectionPersister getCollectionDescriptor();
 
+	/**
+	 * The initializer responsible for the loading
+	 */
 	CollectionInitializer getInitializer();
 
+	/**
+	 * The collection key.
+	 */
 	Serializable getKey();
 
+	/**
+	 * The collection instance being loaded
+	 */
 	PersistentCollection getCollectionInstance();
 
+	/**
+	 * Callback for row loading.  Allows delayed List creation
+	 */
+	void load(Consumer<List> loadingEntryConsumer);
+
+	/**
+	 * Complete the load
+	 */
 	void finishLoading(ExecutionContext executionContext);
 }

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

@@ -6,24 +6,24 @@
  */
 package org.hibernate.sql.results.graph.collection.internal;
 
+import java.util.List;
+
 import org.hibernate.LockMode;
 import org.hibernate.collection.spi.CollectionSemantics;
 import org.hibernate.collection.spi.PersistentCollection;
-import org.hibernate.engine.spi.CollectionEntry;
 import org.hibernate.engine.spi.CollectionKey;
 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.sql.results.graph.collection.CollectionLoadingLogger;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.query.NavigablePath;
-import org.hibernate.sql.exec.spi.ExecutionContext;
-import org.hibernate.sql.results.internal.LoadingCollectionEntryImpl;
 import org.hibernate.sql.results.graph.DomainResultAssembler;
 import org.hibernate.sql.results.graph.FetchParentAccess;
+import org.hibernate.sql.results.graph.collection.CollectionLoadingLogger;
 import org.hibernate.sql.results.graph.collection.LoadingCollectionEntry;
+import org.hibernate.sql.results.internal.LoadingCollectionEntryImpl;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 
 /**
@@ -41,8 +41,8 @@ public abstract class AbstractImmediateCollectionInitializer extends AbstractCol
 
 	// per-row state
 	private PersistentCollection collectionInstance;
+	private LoadingCollectionEntryImpl responsibility;
 	private boolean responsible;
-	private boolean collectionEmpty = true;
 
 	public AbstractImmediateCollectionInitializer(
 			NavigablePath collectionPath,
@@ -110,7 +110,7 @@ public abstract class AbstractImmediateCollectionInitializer extends AbstractCol
 
 			if ( existingLoadingEntry.getInitializer() == this ) {
 				// we are responsible for loading the collection values
-				responsible = true;
+				responsibility = (LoadingCollectionEntryImpl) existingLoadingEntry;
 			}
 			else {
 				// the entity is already being loaded elsewhere
@@ -215,7 +215,7 @@ public abstract class AbstractImmediateCollectionInitializer extends AbstractCol
 			takeResponsibility( rowProcessingState, collectionKey );
 		}
 
-		if ( responsible ) {
+		if ( responsibility != null ) {
 			if ( CollectionLoadingLogger.DEBUG_ENABLED ) {
 				CollectionLoadingLogger.INSTANCE.debugf(
 						"(%s) Responsible for loading collection [%s] : %s",
@@ -238,32 +238,31 @@ public abstract class AbstractImmediateCollectionInitializer extends AbstractCol
 	}
 
 	protected void takeResponsibility(RowProcessingState rowProcessingState, CollectionKey collectionKey) {
+		responsibility = new LoadingCollectionEntryImpl(
+				getCollectionAttributeMapping().getCollectionDescriptor(),
+				this,
+				collectionKey.getKey(),
+				collectionInstance
+		);
 		rowProcessingState.getJdbcValuesSourceProcessingState().registerLoadingCollection(
 				collectionKey,
-				new LoadingCollectionEntryImpl(
+				responsibility
-						getCollectionAttributeMapping().getCollectionDescriptor(),
-						this,
-						collectionKey.getKey(),
-						collectionInstance
-				)
 		);
-		responsible = true;
 	}
 
 	@Override
 	public void initializeInstance(RowProcessingState rowProcessingState) {
-		if ( !responsible ) {
+		if ( responsibility == null ) {
 			return;
 		}
 
-		final PersistenceContext persistenceContext = rowProcessingState.getSession().getPersistenceContext();
-
 		// the LHS key value of the association
 		final CollectionKey collectionKey = resolveCollectionKey( rowProcessingState );
-		// the RHS key value of the association
-		final Object keyCollectionValue = getKeyCollectionValue();
 
-		if ( keyCollectionValue != null ) {
+		// the RHS key value of the association - determines if the row contains an element of the initializing collection
+		final Object collectionValueKey = getKeyCollectionValue();
+
+		if ( collectionValueKey != null ) {
 			// the row contains an element in the collection...
 			if ( CollectionLoadingLogger.DEBUG_ENABLED ) {
 				CollectionLoadingLogger.INSTANCE.debugf(
@@ -274,45 +273,20 @@ public abstract class AbstractImmediateCollectionInitializer extends AbstractCol
 				);
 			}
 
-			readCollectionRow( rowProcessingState );
+			responsibility.load(
-			collectionEmpty = false;
+					loadingState -> readCollectionRow( collectionKey, loadingState, rowProcessingState )
+			);
 		}
 	}
 
-	protected abstract void readCollectionRow(RowProcessingState rowProcessingState);
+	protected abstract void readCollectionRow(CollectionKey collectionKey, List loadingState, RowProcessingState rowProcessingState);
 
 	@Override
 	public void finishUpRow(RowProcessingState rowProcessingState) {
 		super.finishUpRow( rowProcessingState );
 
 		collectionInstance = null;
-		responsible = false;
+		responsibility = null;
-		collectionEmpty = true;
 	}
 
-	@Override
-	public void endLoading(ExecutionContext context) {
-		if ( getParentAccess() == null && collectionEmpty ) {
-			// collection is empty; handle special logic here.
-			final CollectionKey collectionKey = context.getCollectionKey();
-			if ( collectionKey != null ) {
-				// We expected to load a collection with this collection key but we found the collection
-				// contained no results, therefore we need to do the collection init phase here because
-				// the LoadingCollectionEntry won't finalize this for us without at least one row.
-				final PersistenceContext persistenceContext = context.getSession().getPersistenceContext();
-				final PersistentCollection collection = persistenceContext.getCollection( collectionKey );
-
-				if ( ! collection.isInitializing() ) {
-					collection.beforeInitialize( getCollectionAttributeMapping().getCollectionDescriptor(), 0 );
-					collection.beginRead();
-					collection.endRead();
-
-					final CollectionEntry entry = persistenceContext.getCollectionEntry( collection );
-					if ( entry != null ) {
-						entry.postInitialize( collection );
-					}
-				}
-			}
-		}
-	}
 }

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

@@ -6,8 +6,11 @@
  */
 package org.hibernate.sql.results.graph.collection.internal;
 
+import java.util.List;
+
 import org.hibernate.LockMode;
 import org.hibernate.collection.internal.PersistentArrayHolder;
+import org.hibernate.engine.spi.CollectionKey;
 import org.hibernate.internal.log.LoggingHelper;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
@@ -55,12 +58,23 @@ public class ArrayInitializer extends AbstractImmediateCollectionInitializer {
 	}
 
 	@Override
-	protected void readCollectionRow(RowProcessingState rowProcessingState) {
+	protected void readCollectionRow(
+			CollectionKey collectionKey,
+			List loadingState,
+			RowProcessingState rowProcessingState) {
 		int index = (int) listIndexAssembler.assemble( rowProcessingState );
+
 		if ( indexBase != 0 ) {
 			index -= indexBase;
 		}
-		getCollectionInstance().load( index, elementAssembler.assemble( rowProcessingState ) );
+
+		for ( int i = loadingState.size(); i <= index; ++i ) {
+			//noinspection unchecked
+			loadingState.add( i, null );
+		}
+
+		//noinspection unchecked
+		loadingState.set( index, elementAssembler.assemble( rowProcessingState ) );
 	}
 
 	@Override

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

@@ -6,9 +6,12 @@
  */
 package org.hibernate.sql.results.graph.collection.internal;
 
+import java.util.List;
+
 import org.hibernate.LockMode;
 import org.hibernate.collection.internal.PersistentBag;
 import org.hibernate.collection.internal.PersistentIdentifierBag;
+import org.hibernate.engine.spi.CollectionKey;
 import org.hibernate.internal.log.LoggingHelper;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
@@ -42,16 +45,21 @@ public class BagInitializer extends AbstractImmediateCollectionInitializer {
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
+	protected void readCollectionRow(
-	protected void readCollectionRow(RowProcessingState rowProcessingState) {
+			CollectionKey collectionKey,
+			List loadingState,
+			RowProcessingState rowProcessingState) {
 		if ( collectionIdAssembler != null ) {
-			( (PersistentIdentifierBag) getCollectionInstance() ).load(
+			final Object[] row = new Object[2];
-					collectionIdAssembler.assemble( rowProcessingState ),
+			row[0] = collectionIdAssembler.assemble( rowProcessingState );
-					elementAssembler.assemble( rowProcessingState )
+			row[1] = elementAssembler.assemble( rowProcessingState );
-			);
+
+			//noinspection unchecked
+			loadingState.add( row );
 		}
 		else {
-			( (PersistentBag) getCollectionInstance() ).load( elementAssembler.assemble( rowProcessingState ) );
+			//noinspection unchecked
+			loadingState.add( elementAssembler.assemble( rowProcessingState ) );
 		}
 	}
 

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

@@ -6,8 +6,11 @@
  */
 package org.hibernate.sql.results.graph.collection.internal;
 
+import java.util.List;
+
 import org.hibernate.LockMode;
 import org.hibernate.collection.internal.PersistentList;
+import org.hibernate.engine.spi.CollectionKey;
 import org.hibernate.internal.log.LoggingHelper;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
@@ -49,13 +52,23 @@ public class ListInitializer extends AbstractImmediateCollectionInitializer {
 	}
 
 	@Override
-	protected void readCollectionRow(RowProcessingState rowProcessingState) {
+	protected void readCollectionRow(
+			CollectionKey collectionKey,
+			List loadingState,
+			RowProcessingState rowProcessingState) {
 		int index = (int) listIndexAssembler.assemble( rowProcessingState );
-		getCollectionAttributeMapping().getIndexMetadata().getIndexDescriptor();
+
 		if ( listIndexBase != 0 ) {
 			index -= listIndexBase;
 		}
-		getCollectionInstance().load( index, elementAssembler.assemble( rowProcessingState ) );
+
+		for ( int i = loadingState.size(); i <= index; ++i ) {
+			//noinspection unchecked
+			loadingState.add( i, null );
+		}
+
+		//noinspection unchecked
+		loadingState.set( index, elementAssembler.assemble( rowProcessingState ) );
 	}
 
 	@Override

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

@@ -6,8 +6,11 @@
  */
 package org.hibernate.sql.results.graph.collection.internal;
 
+import java.util.List;
+
 import org.hibernate.LockMode;
 import org.hibernate.collection.internal.PersistentMap;
+import org.hibernate.engine.spi.CollectionKey;
 import org.hibernate.internal.log.LoggingHelper;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
@@ -48,11 +51,16 @@ public class MapInitializer extends AbstractImmediateCollectionInitializer {
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
+	protected void readCollectionRow(
-	protected void readCollectionRow(RowProcessingState rowProcessingState) {
+			CollectionKey collectionKey,
-		getCollectionInstance().load(
+			List loadingState,
-				mapKeyAssembler.assemble( rowProcessingState ),
+			RowProcessingState rowProcessingState) {
-				mapValueAssembler.assemble( rowProcessingState )
+		//noinspection unchecked
+		loadingState.add(
+				new Object[] {
+						mapKeyAssembler.assemble( rowProcessingState ),
+						mapValueAssembler.assemble( rowProcessingState )
+				}
 		);
 	}
 

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

@@ -6,8 +6,11 @@
  */
 package org.hibernate.sql.results.graph.collection.internal;
 
+import java.util.List;
+
 import org.hibernate.LockMode;
 import org.hibernate.collection.internal.PersistentSet;
+import org.hibernate.engine.spi.CollectionKey;
 import org.hibernate.internal.log.LoggingHelper;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
@@ -40,8 +43,12 @@ public class SetInitializer extends AbstractImmediateCollectionInitializer {
 	}
 
 	@Override
-	protected void readCollectionRow(RowProcessingState rowProcessingState) {
+	protected void readCollectionRow(
-		getCollectionInstance().load( elementAssembler.assemble( rowProcessingState ) );
+			CollectionKey collectionKey,
+			List loadingState,
+			RowProcessingState rowProcessingState) {
+		//noinspection unchecked
+		loadingState.add( elementAssembler.assemble( rowProcessingState ) );
 	}
 
 	@Override

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

@@ -7,7 +7,11 @@
 package org.hibernate.sql.results.internal;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
 
+import org.hibernate.collection.internal.PersistentArrayHolder;
 import org.hibernate.collection.spi.PersistentCollection;
 import org.hibernate.engine.spi.BatchFetchQueue;
 import org.hibernate.engine.spi.CollectionEntry;
@@ -29,6 +33,8 @@ public class LoadingCollectionEntryImpl implements LoadingCollectionEntry {
 	private final Object key;
 	private final PersistentCollection collectionInstance;
 
+	private List loadingState;
+
 	public LoadingCollectionEntryImpl(
 			CollectionPersister collectionDescriptor,
 			CollectionInitializer initializer,
@@ -64,13 +70,22 @@ public class LoadingCollectionEntryImpl implements LoadingCollectionEntry {
 	}
 
 	@Override
-	public String toString() {
+	public void load(Consumer<List> loadingEntryConsumer) {
-		return getClass().getSimpleName() + "(" + getCollectionDescriptor().getNavigableRole().getFullPath() + "#" + getKey() + ")";
+		if ( loadingState == null ) {
+			loadingState = new ArrayList();
+		}
+		loadingEntryConsumer.accept( loadingState );
 	}
 
 	@Override public void finishLoading(ExecutionContext executionContext) {
+		collectionInstance.injectLoadedState(
+				getCollectionDescriptor().getAttributeMapping(),
+				loadingState
+		);
+
 		collectionInstance.endRead();
 
+
 		final SharedSessionContractImplementor session = executionContext.getSession();
 		final PersistenceContext persistenceContext = session.getPersistenceContext();
 		final CollectionPersister collectionDescriptor = getCollectionDescriptor();
@@ -97,4 +112,9 @@ public class LoadingCollectionEntryImpl implements LoadingCollectionEntry {
 		// todo (6.0) : there is other logic still needing to be implemented here.  caching, etc
 		// 		see org.hibernate.engine.loading.internal.CollectionLoadContext#endLoadingCollection in 5.x
 	}
+
+	@Override
+	public String toString() {
+		return getClass().getSimpleName() + "(" + getCollectionDescriptor().getNavigableRole().getFullPath() + "#" + getKey() + ")";
+	}
 }