Implement support for delayed collections with FKs based on non-primary keys

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

@@ -1003,7 +1003,18 @@ public class MappingModelCreationHelper {
 			return;
 		}
 
-		final ModelPart fkTarget = referencedEntityDescriptor.getIdentifierMapping();
+		final ModelPart fkTarget;
+		if ( bootValueMapping.isReferenceToPrimaryKey() ) {
+			fkTarget = referencedEntityDescriptor.getIdentifierMapping();
+		}
+		else {
+			// TODO: need some kind of virtual model part for this
+//			fkTarget = attributeMapping;//bootValueMapping.;
+			throw new NotYetImplementedFor6Exception(
+					"Support for non-pk foreign-keys not yet implemented: " +
+							bootProperty.getPersistentClass().getEntityName() + " -> " + bootProperty.getName()
+			);
+		}
 
 		if ( fkTarget instanceof BasicValuedModelPart ) {
 			final BasicValuedModelPart simpleFkTarget = (BasicValuedModelPart) fkTarget;
@@ -1049,7 +1060,7 @@ public class MappingModelCreationHelper {
 		}
 		else {
 			throw new NotYetImplementedFor6Exception(
-					"Support for" + fkTarget.getClass() + " foreign-keys not yet implemented: " +
+					"Support for " + fkTarget.getClass() + " foreign-keys not yet implemented: " +
 							bootProperty.getPersistentClass().getEntityName() + " -> " + bootProperty.getName()
 			);
 		}

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

@@ -518,10 +518,19 @@ public class PluralAttributeMappingImpl
 			return new SelectEagerCollectionFetch( fetchablePath, this, fetchParent );
 		}
 
+		final FromClauseAccess fromClauseAccess = sqlAstCreationState.getFromClauseAccess();
+		final TableGroup fetchParentTableGroup = fromClauseAccess.getTableGroup( fetchParent.getNavigablePath() );
+		final DomainResult fkResult = getKeyDescriptor().createDomainResult(
+				fetchablePath,
+				fetchParentTableGroup,
+				false,
+				creationState
+		);
 		return new DelayedCollectionFetch(
 				fetchablePath,
 				this,
-				fetchParent
+				fetchParent,
+				fkResult
 		);
 	}
 

hibernate-core/src/main/java/org/hibernate/query/sqm/sql/BaseSqmToSqlAstConverter.java

@@ -1406,9 +1406,21 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 				final int sqmPosition = (Integer) literal;
 				final List<SqlSelection> selections = currentSqlSelectionCollector().getSelections( sqmPosition );
 				final List<Expression> expressions = new ArrayList<>( selections.size() );
-				for ( SqlSelection selection : selections ) {
+				OUTER: for ( int i = 0; i < selections.size(); i++ ) {
+					final SqlSelection selection = selections.get( i );
+					// We skip duplicate selections which can occur when grouping/ordering by an entity alias.
+					// Duplication happens because the primary key of an entity usually acts as FK target of collections
+					// which is, just like the identifier itself, also registered as selection
+					for ( int j = 0; j < i; j++ ) {
+						if ( selections.get( j ) == selection ) {
+							continue OUTER;
+						}
+					}
 					expressions.add( new SqlSelectionExpression( selection ) );
 				}
+				if ( expressions.size() == 1 ) {
+					return expressions.get( 0 );
+				}
 				return new SqlTuple( expressions, null );
 			}
 		}

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

@@ -11,6 +11,8 @@ import java.util.function.Consumer;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
 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.FetchParentAccess;
 import org.hibernate.sql.results.graph.Initializer;
 import org.hibernate.sql.results.graph.collection.CollectionInitializer;
@@ -23,13 +25,22 @@ public class DelayedCollectionAssembler extends AbstractCollectionAssembler {
 			NavigablePath fetchPath,
 			PluralAttributeMapping fetchedMapping,
 			FetchParentAccess parentAccess,
+			DomainResult fkResult,
 			AssemblerCreationState creationState) {
 		super(
 				fetchedMapping,
 				() -> (CollectionInitializer) creationState.resolveInitializer(
 						fetchPath,
 						fetchedMapping,
-						() -> new DelayedCollectionInitializer( fetchPath, fetchedMapping, parentAccess )
+						() -> {
+							final DomainResultAssembler fkAssembler = fkResult.createResultAssembler( creationState );
+							return new DelayedCollectionInitializer(
+									fetchPath,
+									fetchedMapping,
+									parentAccess,
+									fkAssembler
+							);
+						}
 				)
 
 		);

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

@@ -12,6 +12,7 @@ import org.hibernate.engine.FetchTiming;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.NavigablePath;
 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.FetchParent;
 import org.hibernate.sql.results.graph.FetchParentAccess;
@@ -22,11 +23,16 @@ import org.hibernate.type.descriptor.java.JavaTypeDescriptor;
  * @author Steve Ebersole
  */
 public class DelayedCollectionFetch extends CollectionFetch {
+
+	private final DomainResult fkResult;
+
 	public DelayedCollectionFetch(
 			NavigablePath fetchedPath,
 			PluralAttributeMapping fetchedAttribute,
-			FetchParent fetchParent) {
+			FetchParent fetchParent,
+			DomainResult fkResult) {
 		super( fetchedPath, fetchedAttribute, fetchParent );
+		this.fkResult = fkResult;
 	}
 
 	@Override
@@ -37,6 +43,7 @@ public class DelayedCollectionFetch extends CollectionFetch {
 				getNavigablePath(),
 				getFetchedMapping(),
 				parentAccess,
+				fkResult,
 				creationState
 		);
 	}

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

@@ -8,6 +8,7 @@ package org.hibernate.sql.results.graph.collection.internal;
 
 import org.hibernate.collection.spi.CollectionSemantics;
 import org.hibernate.collection.spi.PersistentCollection;
+import org.hibernate.engine.spi.CollectionKey;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.log.LoggingHelper;
@@ -15,8 +16,10 @@ import org.hibernate.metamodel.CollectionClassification;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.results.graph.DomainResultAssembler;
 import org.hibernate.sql.results.graph.FetchParentAccess;
 import org.hibernate.sql.results.graph.collection.LoadingCollectionEntry;
+import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingOptions;
 import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
 
 /**
@@ -24,11 +27,67 @@ import org.hibernate.sql.results.jdbc.spi.RowProcessingState;
  */
 public class DelayedCollectionInitializer extends AbstractCollectionInitializer {
 
+	/**
+	 * refers to the collection's container value - which collection-key?
+	 */
+	private final DomainResultAssembler keyContainerAssembler;
+
 	public DelayedCollectionInitializer(
 			NavigablePath fetchedPath,
 			PluralAttributeMapping fetchedMapping,
-			FetchParentAccess parentAccess) {
+			FetchParentAccess parentAccess,
+			DomainResultAssembler keyContainerAssembler) {
 		super( fetchedPath, fetchedMapping, parentAccess );
+		this.keyContainerAssembler = keyContainerAssembler;
+	}
+
+	@Override
+	public void resolveKey(RowProcessingState rowProcessingState) {
+		if ( collectionKey != null ) {
+			// already resolved
+			return;
+		}
+
+		final CollectionKey loadingKey = rowProcessingState.getCollectionKey();
+		if ( loadingKey != null ) {
+			collectionKey = loadingKey;
+			return;
+		}
+
+		final JdbcValuesSourceProcessingOptions processingOptions = rowProcessingState.getJdbcValuesSourceProcessingState()
+				.getProcessingOptions();
+
+		final Object keyContainerValue = keyContainerAssembler.assemble(
+				rowProcessingState,
+				processingOptions
+		);
+		if ( keyContainerValue != null ) {
+			this.collectionKey = new CollectionKey(
+					collectionAttributeMapping.getCollectionDescriptor(),
+					keyContainerValue
+			);
+
+			// TODO: This fails e.g. EagerCollectionLazyKeyManyToOneTest because Order$Id#customer is null
+			//   which is required for the hash code. Is this being null at this point a bug?
+//			if ( CollectionLoadingLogger.DEBUG_ENABLED ) {
+//				CollectionLoadingLogger.INSTANCE.debugf(
+//						"(%s) Current row collection key : %s",
+//						DelayedCollectionInitializer.class.getSimpleName(),
+//						LoggingHelper.toLoggableString( getNavigablePath(), this.collectionKey.getKey() )
+//				);
+//			}
+			parentAccess.registerResolutionListener( owner -> collectionInstance.setOwner( owner ) );
+		}
+		else {
+			final Object parentKey = parentAccess.getParentKey();
+			if ( parentKey != null ) {
+				this.collectionKey = new CollectionKey(
+						collectionAttributeMapping.getCollectionDescriptor(),
+						parentKey
+				);
+				parentAccess.registerResolutionListener( owner -> collectionInstance.setOwner( owner ) );
+			}
+		}
 	}
 
 	@Override