Fix PluralAttribute manyToMany FK target part determination

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

@@ -983,20 +983,33 @@ public class MappingModelCreationHelper {
 			Value bootValueMapping,
 			Dialect dialect,
 			MappingModelCreationProcess creationProcess) {
-		final List<String> keyColumnExpressions = new ArrayList<>(bootValueMapping.getColumnSpan());
-		bootValueMapping.getColumnIterator().forEachRemaining(
-				column ->
-						keyColumnExpressions.add( column.getText( dialect ) ) );
-
 		final List<String> mappedColumnExpressions = fkTarget.getMappedColumnExpressions();
 		final List<String> targetColumnExpressions = new ArrayList<>( mappedColumnExpressions.size() );
 		mappedColumnExpressions.forEach(
 				column ->
 						targetColumnExpressions.add( column ) );
 
+		final List<String> keyColumnExpressions;
+		Table keyTableExpression;
+		if ( bootValueMapping instanceof Collection ) {
+			final Collection collectioBootValueMapping = (Collection) bootValueMapping;
+			keyTableExpression = collectioBootValueMapping.getCollectionTable();
+			final KeyValue key = collectioBootValueMapping.getKey();
+			keyColumnExpressions = new ArrayList<>( key.getColumnSpan() );
+			key.getColumnIterator().forEachRemaining(
+					column ->
+							keyColumnExpressions.add( column.getText( dialect ) ) );
+		}
+		else {
+			keyTableExpression = bootValueMapping.getTable();
+			keyColumnExpressions = new ArrayList<>( bootValueMapping.getColumnSpan() );
+			bootValueMapping.getColumnIterator().forEachRemaining(
+					column ->
+							keyColumnExpressions.add( column.getText( dialect ) ) );
+		}
 		return new EmbeddedForeignKeyDescriptor(
 				(EmbeddedIdentifierMappingImpl) fkTarget,
-				getTableIdentifierExpression( bootValueMapping.getTable(), creationProcess ),
+				getTableIdentifierExpression( keyTableExpression, creationProcess ),
 				keyColumnExpressions,
 				fkTarget.getContainingTableExpression(),
 				targetColumnExpressions,

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

@@ -40,7 +40,9 @@ import org.hibernate.metamodel.mapping.ordering.OrderByFragment;
 import org.hibernate.metamodel.mapping.ordering.OrderByFragmentTranslator;
 import org.hibernate.metamodel.mapping.ordering.TranslationContext;
 import org.hibernate.metamodel.model.domain.NavigableRole;
+import org.hibernate.persister.collection.AbstractCollectionPersister;
 import org.hibernate.persister.collection.CollectionPersister;
+import org.hibernate.persister.entity.AbstractEntityPersister;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.persister.entity.Joinable;
 import org.hibernate.property.access.spi.PropertyAccess;
@@ -234,9 +236,15 @@ public class PluralAttributeMappingImpl extends AbstractAttributeMapping impleme
 						if ( collectionDescriptor.getElementType() instanceof EntityType ) {
 							final EntityType elementEntityType = (EntityType) collectionDescriptor.getElementType();
 							associatedEntityDescriptor = creationProcess.getEntityPersister( elementEntityType.getAssociatedEntityName() );
-							fkTargetPart = elementEntityType.isReferenceToPrimaryKey()
+							if ( ( (AbstractEntityPersister) associatedEntityDescriptor ).getTableName()
-									? associatedEntityDescriptor.getIdentifierMapping()
+									.equals( ( (AbstractCollectionPersister) collectionDescriptor ).getTableName() ) ) {
-									: associatedEntityDescriptor.findSubPart( elementEntityType.getRHSUniqueKeyPropertyName() );
+								fkTargetPart = creationProcess
+										.getEntityPersister( bootDescriptor.getOwner().getEntityName() )
+										.getIdentifierMapping();
+							}
+							else {
+								fkTargetPart = associatedEntityDescriptor.getIdentifierMapping();
+							}
 							fkBootDescriptorSource = bootDescriptor.getElement();
 						}
 						else {

hibernate-core/src/main/java/org/hibernate/metamodel/model/domain/internal/BagAttributeImpl.java

@@ -8,10 +8,8 @@ package org.hibernate.metamodel.model.domain.internal;
 
 import java.util.Collection;
 
-import org.hibernate.NotYetImplementedFor6Exception;
 import org.hibernate.metamodel.internal.MetadataContext;
 import org.hibernate.metamodel.model.domain.BagPersistentAttribute;
-import org.hibernate.query.sqm.SqmPathSource;
 import org.hibernate.query.hql.spi.SqmCreationState;
 import org.hibernate.query.sqm.tree.SqmJoinType;
 import org.hibernate.query.sqm.tree.domain.SqmBagJoin;
@@ -34,11 +32,6 @@ class BagAttributeImpl<X, E>
 		return CollectionType.COLLECTION;
 	}
 
-	@Override
-	public SqmPathSource<?> findSubPathSource(String name) {
-		throw new NotYetImplementedFor6Exception();
-	}
-
 	@Override
 	public SqmAttributeJoin createSqmJoin(
 			SqmFrom lhs,

hibernate-core/src/main/java/org/hibernate/metamodel/model/domain/internal/MapAttributeImpl.java

@@ -8,7 +8,6 @@ package org.hibernate.metamodel.model.domain.internal;
 
 import java.util.Map;
 
-import org.hibernate.NotYetImplementedFor6Exception;
 import org.hibernate.metamodel.internal.MetadataContext;
 import org.hibernate.metamodel.model.domain.MapPersistentAttribute;
 import org.hibernate.metamodel.model.domain.SimpleDomainType;
@@ -66,11 +65,6 @@ class MapAttributeImpl<X, K, V> extends AbstractPluralAttribute<X, Map<K, V>, V>
 		return getKeyType();
 	}
 
-	@Override
-	public SqmPathSource<?> findSubPathSource(String name) {
-		throw new NotYetImplementedFor6Exception();
-	}
-
 	@Override
 	public SqmAttributeJoin createSqmJoin(
 			SqmFrom lhs, SqmJoinType joinType, String alias, boolean fetched, SqmCreationState creationState) {