HHH-15349 Fix rendering of EntityValuedPathInterpretation when comparing different model parts

documentation/src/test/java/org/hibernate/userguide/associations/NotFoundTest.java

@@ -2,6 +2,8 @@ package org.hibernate.userguide.associations;
 
 import java.io.Serializable;
 import java.util.List;
+import java.util.Map;
+
 import jakarta.persistence.Entity;
 import jakarta.persistence.FetchType;
 import jakarta.persistence.GeneratedValue;
@@ -14,6 +16,7 @@ import org.hibernate.annotations.NotFound;
 import org.hibernate.annotations.NotFoundAction;
 import org.hibernate.orm.test.jpa.BaseEntityManagerFunctionalTestCase;
 
+import org.hibernate.testing.jdbc.SQLStatementInterceptor;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -29,6 +32,13 @@ import static org.junit.Assert.assertNull;
  */
 public class NotFoundTest extends BaseEntityManagerFunctionalTestCase {
 
+	private SQLStatementInterceptor sqlStatementInterceptor;
+
+	@Override
+	protected void addConfigOptions(Map options) {
+		sqlStatementInterceptor = new SQLStatementInterceptor( options );
+	}
+
 	@Override
 	protected Class<?>[] getAnnotatedClasses() {
 		return new Class<?>[] {
@@ -111,19 +121,25 @@ public class NotFoundTest extends BaseEntityManagerFunctionalTestCase {
 		breakForeignKey();
 
 		inTransaction( entityManagerFactory(), (entityManager) -> {
+			sqlStatementInterceptor.clear();
 			//tag::associations-not-found-fk-function-example[]
-			final List<Person> nullResults = entityManager
+			final List<String> nullResults = entityManager
-					.createSelectionQuery( "from Person p where fk( p.city ) is null", Person.class )
+					.createSelectionQuery( "select p.name from Person p where fk( p.city ) is null", String.class )
 					.list();
 
 			assertThat( nullResults ).isEmpty();
 
-			final List<Person> nonNullResults = entityManager
+			final List<String> nonNullResults = entityManager
-					.createSelectionQuery( "from Person p where fk( p.city ) is not null", Person.class )
+					.createSelectionQuery( "select p.name from Person p where fk( p.city ) is not null", String.class )
 					.list();
 			assertThat( nonNullResults ).hasSize( 1 );
-			assertThat( nonNullResults.get( 0 ).getName() ).isEqualTo( "John Doe" );
+			assertThat( nonNullResults.get( 0 ) ).isEqualTo( "John Doe" );
 			//end::associations-not-found-fk-function-example[]
+
+			// In addition, make sure that the two executed queries do not create a join
+			assertThat( sqlStatementInterceptor.getQueryCount() ).isEqualTo( 2 );
+			assertThat( sqlStatementInterceptor.getSqlQueries().get( 0 ) ).doesNotContain( " join " );
+			assertThat( sqlStatementInterceptor.getSqlQueries().get( 1 ) ).doesNotContain( " join " );
 		} );
 	}
 

hibernate-core/src/main/java/org/hibernate/cfg/BinderHelper.java

@@ -135,6 +135,10 @@ public class BinderHelper {
 			String syntheticPropertyName =
 					"_" + associatedClass.getEntityName().replace('.', '_') +
 					"_" + columns[0].getPropertyName().replace('.', '_');
+			if ( inverse ) {
+				// Use a different name for inverse synthetic properties to avoid duplicate properties for self-referencing models
+				syntheticPropertyName += "_inverse";
+			}
 			//find properties associated to a certain column
 			Object columnOwner = findColumnOwner( ownerEntity, columns[0].getReferencedColumn(), context );
 			List<Property> properties = findPropertiesByColumns( columnOwner, columns, context );

hibernate-core/src/main/java/org/hibernate/cfg/annotations/CollectionBinder.java

@@ -2271,7 +2271,7 @@ public abstract class CollectionBinder {
 			}
 			String referencedPropertyName =
 					buildingContext.getMetadataCollector().getPropertyReferencedAssociation(
-							"inverse__" + referencedEntity.getEntityName(), mappedBy
+							referencedEntity.getEntityName(), mappedBy
 					);
 			if ( referencedPropertyName != null ) {
 				//TODO always a many to one?

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

@@ -27,6 +27,10 @@ public interface EntityAssociationMapping extends ModelPart, Association, TableG
 	 */
 	ModelPart getKeyTargetMatchPart();
 
+	boolean isReferenceToPrimaryKey();
+
+	boolean isFkOptimizationAllowed();
+
 	@Override
 	default boolean incrementFetchDepth(){
 		return true;

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

@@ -50,6 +50,15 @@ public interface ForeignKeyDescriptor extends VirtualModelPart, ValueMapping {
 
 	String getTargetTable();
 
+	default String getTable(Nature nature) {
+		if ( nature == Nature.KEY ) {
+			return getKeyTable();
+		}
+		else {
+			return getTargetTable();
+		}
+	}
+
 	ModelPart getKeyPart();
 
 	ModelPart getTargetPart();

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

@@ -341,9 +341,14 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor {
 			SqlAstCreationContext creationContext) {
 		final TableReference lhsTableReference = targetSideTableGroup.resolveTableReference(
 				targetSideTableGroup.getNavigablePath(),
-				targetTable
+				targetTable,
+				false
+		);
+		final TableReference rhsTableKeyReference = keySideTableGroup.resolveTableReference(
+				null,
+				keyTable,
+				false
 		);
-		final TableReference rhsTableKeyReference = keySideTableGroup.resolveTableReference( keyTable );
 
 		return generateJoinPredicate(
 				lhsTableReference,
@@ -459,27 +464,6 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor {
 		return true;
 	}
 
-	protected TableReference getTableReference(TableGroup lhs, TableGroup tableGroup, String table) {
-		TableReference tableReference = lhs.getPrimaryTableReference().resolveTableReference( table );
-		if ( tableReference != null ) {
-			return tableReference;
-		}
-		tableReference = tableGroup.getPrimaryTableReference().resolveTableReference( table );
-		if ( tableReference != null ) {
-			return tableReference;
-		}
-
-		tableReference = lhs.resolveTableReference(
-				lhs.getNavigablePath().append( getNavigableRole().getNavigableName() ),
-				table
-		);
-		if ( tableReference != null ) {
-			return tableReference;
-		}
-
-		throw new IllegalStateException( "Could not resolve binding for table `" + table + "`" );
-	}
-
 	@Override
 	public int visitKeySelectables(int offset, SelectableConsumer consumer) {
 		return keySelectableMappings.forEachSelectable( offset, consumer );

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

@@ -600,6 +600,20 @@ public class EntityCollectionPart
 				: ForeignKeyDescriptor.Nature.KEY;
 	}
 
+	@Override
+	public boolean isReferenceToPrimaryKey() {
+		return fkDescriptor.getSide( getSideNature().inverse() ).getModelPart() instanceof EntityIdentifierMapping;
+	}
+
+	@Override
+	public boolean isFkOptimizationAllowed() {
+		return !collectionDescriptor.isOneToMany();
+	}
+
+	public CollectionPersister getCollectionDescriptor() {
+		return collectionDescriptor;
+	}
+
 	@Override
 	public FetchStyle getStyle() {
 		return FetchStyle.JOIN;
@@ -710,7 +724,9 @@ public class EntityCollectionPart
 						return false;
 					}
 
-					return targetKeyPropertyNames.contains( relativePath );
+					// Empty relative path means the navigable paths are equal,
+					// in which case we allow resolving the parent table group
+					return relativePath.isEmpty() || targetKeyPropertyNames.contains( relativePath );
 				},
 				this,
 				explicitSourceAlias,

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

@@ -316,10 +316,13 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 			SqlAstCreationContext creationContext) {
 		final TableReference lhsTableReference = targetSideTableGroup.resolveTableReference(
 				targetSideTableGroup.getNavigablePath(),
-				targetSide.getModelPart().getContainingTableExpression()
+				targetSide.getModelPart().getContainingTableExpression(),
+				false
 		);
 		final TableReference rhsTableKeyReference = keySideTableGroup.resolveTableReference(
-				keySide.getModelPart().getContainingTableExpression()
+				null,
+				keySide.getModelPart().getContainingTableExpression(),
+				false
 		);
 
 		return generateJoinPredicate(
@@ -352,23 +355,6 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 				|| ( lhs.equals( targetExpression ) && rhs.equals( keyExpression ) );
 	}
 
-	protected TableReference getTableReference(TableGroup lhs, TableGroup tableGroup, String table) {
-		final NavigablePath navigablePath = lhs.getNavigablePath().append( getTargetPart().getFetchableName() );
-		if ( lhs.getPrimaryTableReference().getTableReference( navigablePath, table ) != null ) {
-			return lhs.getPrimaryTableReference();
-		}
-		else if ( tableGroup.getPrimaryTableReference().getTableReference( navigablePath, table ) != null ) {
-			return tableGroup.getPrimaryTableReference();
-		}
-
-		final TableReference tableReference = lhs.resolveTableReference( navigablePath, table );
-		if ( tableReference != null ) {
-			return tableReference;
-		}
-
-		throw new IllegalStateException( "Could not resolve binding for table `" + table + "`" );
-	}
-
 	@Override
 	public MappingType getPartMappingType() {
 		return targetSide.getModelPart().getMappedType();

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

@@ -703,10 +703,16 @@ public class ToOneAttributeMapping
 		return sideNature;
 	}
 
+	@Override
 	public boolean isReferenceToPrimaryKey() {
 		return foreignKeyDescriptor.getSide( sideNature.inverse() ).getModelPart() instanceof EntityIdentifierMapping;
 	}
 
+	@Override
+	public boolean isFkOptimizationAllowed() {
+		return canUseParentTableGroup;
+	}
+
 	public String getReferencedPropertyName() {
 		return referencedPropertyName;
 	}
@@ -1576,7 +1582,9 @@ public class ToOneAttributeMapping
 											return false;
 										}
 
-										return targetKeyPropertyNames.contains( relativePath );
+										// Empty relative path means the navigable paths are equal,
+										// in which case we allow resolving the parent table group
+										return relativePath.isEmpty() || targetKeyPropertyNames.contains( relativePath );
 									}
 							),
 							null
@@ -1606,23 +1614,31 @@ public class ToOneAttributeMapping
 
 		final TableReference lhsTableReference = lhs.resolveTableReference( navigablePath, identifyingColumnsTableExpression );
 
-		lazyTableGroup.setTableGroupInitializerCallback( (tableGroup) -> join.applyPredicate(
+		lazyTableGroup.setTableGroupInitializerCallback(
+				tableGroup -> {
+					join.applyPredicate(
 							foreignKeyDescriptor.generateJoinPredicate(
-						sideNature == ForeignKeyDescriptor.Nature.TARGET ? lhsTableReference : tableGroup.getPrimaryTableReference(),
+									sideNature == ForeignKeyDescriptor.Nature.TARGET ?
-						sideNature == ForeignKeyDescriptor.Nature.TARGET ? tableGroup.getPrimaryTableReference() : lhsTableReference,
+											lhsTableReference :
+											tableGroup.getPrimaryTableReference(),
+									sideNature == ForeignKeyDescriptor.Nature.TARGET ?
+											tableGroup.getPrimaryTableReference() :
+											lhsTableReference,
 									sqlExpressionResolver,
 									creationContext
 							)
-		) );
+					);
 
 					if ( hasNotFoundAction() ) {
 						getAssociatedEntityMappingType().applyWhereRestrictions(
 								join::applyPredicate,
-					lazyTableGroup.getTableGroup(),
+								tableGroup,
 								true,
 								null
 						);
 					}
+				}
+		);
 
 		return join;
 	}
@@ -1697,7 +1713,9 @@ public class ToOneAttributeMapping
 						return false;
 					}
 
-					return targetKeyPropertyNames.contains( relativePath );
+					// Empty relative path means the navigable paths are equal,
+					// in which case we allow resolving the parent table group
+					return relativePath.isEmpty() || targetKeyPropertyNames.contains( relativePath );
 				},
 				tableGroupProducer,
 				explicitSourceAlias,

hibernate-core/src/main/java/org/hibernate/query/sqm/internal/SqmMappingModelHelper.java

@@ -145,16 +145,16 @@ public class SqmMappingModelHelper {
 			MappingMetamodel domainModel,
 			Function<NavigablePath,TableGroup> tableGroupLocator) {
 
-		if ( sqmPath instanceof SqmTreatedPath ) {
+		if ( sqmPath instanceof SqmTreatedPath<?, ?> ) {
-			final SqmTreatedPath treatedPath = (SqmTreatedPath) sqmPath;
+			final SqmTreatedPath<?, ?> treatedPath = (SqmTreatedPath<?, ?>) sqmPath;
-			final EntityDomainType treatTargetType = treatedPath.getTreatTarget();
+			final EntityDomainType<?> treatTargetType = treatedPath.getTreatTarget();
 			return domainModel.findEntityDescriptor( treatTargetType.getHibernateEntityName() );
 		}
 
 		// see if the LHS is treated
-		if ( sqmPath.getLhs() instanceof SqmTreatedPath ) {
+		if ( sqmPath.getLhs() instanceof SqmTreatedPath<?, ?> ) {
-			final SqmTreatedPath treatedPath = (SqmTreatedPath) sqmPath.getLhs();
+			final SqmTreatedPath<?, ?> treatedPath = (SqmTreatedPath<?, ?>) sqmPath.getLhs();
-			final EntityDomainType treatTargetType = treatedPath.getTreatTarget();
+			final EntityDomainType<?> treatTargetType = treatedPath.getTreatTarget();
 			final EntityPersister container = domainModel.findEntityDescriptor( treatTargetType.getHibernateEntityName() );
 
 			return container.findSubPart( sqmPath.getNavigablePath().getLocalName(), container );

hibernate-core/src/main/java/org/hibernate/query/sqm/internal/SqmUtil.java

@@ -28,6 +28,7 @@ import org.hibernate.metamodel.mapping.ConvertibleModelPart;
 import org.hibernate.metamodel.mapping.EntityAssociationMapping;
 import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.JdbcMapping;
 import org.hibernate.metamodel.mapping.MappingModelExpressible;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
@@ -350,6 +351,10 @@ public class SqmUtil {
 		if ( parameterType == null ) {
 			throw new SqlTreeCreationException( "Unable to interpret mapping-model type for Query parameter : " + domainParam );
 		}
+		else if ( parameterType instanceof PluralAttributeMapping ) {
+			// Default to the collection element
+			parameterType = ( (PluralAttributeMapping) parameterType ).getElementDescriptor();
+		}
 
 		if ( parameterType instanceof EntityIdentifierMapping ) {
 			final EntityIdentifierMapping identifierMapping = (EntityIdentifierMapping) parameterType;
@@ -368,6 +373,13 @@ public class SqmUtil {
 		}
 		else if ( parameterType instanceof EntityAssociationMapping ) {
 			EntityAssociationMapping association = (EntityAssociationMapping) parameterType;
+			if ( association.getSideNature() == ForeignKeyDescriptor.Nature.TARGET ) {
+				// If the association is the target, we must use the identifier of the EntityMappingType
+				bindValue = association.getAssociatedEntityMappingType().getIdentifierMapping()
+						.getIdentifier( bindValue );
+				parameterType = association.getAssociatedEntityMappingType().getIdentifierMapping();
+			}
+			else {
 				bindValue = association.getForeignKeyDescriptor().getAssociationKeyFromSide(
 						bindValue,
 						association.getSideNature().inverse(),
@@ -375,10 +387,6 @@ public class SqmUtil {
 				);
 				parameterType = association.getForeignKeyDescriptor();
 			}
-		else if ( parameterType instanceof PluralAttributeMapping ) {
-			// we'd expect the values to refer to the collection element
-			// for now, let's blow up and see where this happens and fix the specifics...
-			throw new NotYetImplementedFor6Exception( "Binding parameters whose inferred type comes from plural attribute not yet implemented" );
 		}
 
 		int offset = jdbcParameterBindings.registerParametersForEachJdbcValue(

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

@@ -332,6 +332,7 @@ import org.hibernate.sql.ast.tree.from.SyntheticVirtualTableGroup;
 import org.hibernate.sql.ast.tree.from.TableGroup;
 import org.hibernate.sql.ast.tree.from.TableGroupJoin;
 import org.hibernate.sql.ast.tree.from.TableGroupJoinProducer;
+import org.hibernate.sql.ast.tree.from.TableGroupProducer;
 import org.hibernate.sql.ast.tree.from.TableReference;
 import org.hibernate.sql.ast.tree.from.VirtualTableGroup;
 import org.hibernate.sql.ast.tree.insert.InsertStatement;
@@ -1080,8 +1081,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 					rootTableGroup
 			);
 
-			if ( !rootTableGroup.getTableReferenceJoins().isEmpty()
+			if ( hasJoins( rootTableGroup ) ) {
-					|| !rootTableGroup.getTableGroupJoins().isEmpty() ) {
 				throw new SemanticException( "Not expecting multiple table references for an SQM INSERT-SELECT" );
 			}
 		}
@@ -1109,6 +1109,24 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		return insertStatement;
 	}
 
+	private static boolean hasJoins(TableGroup rootTableGroup) {
+		if ( !rootTableGroup.getTableReferenceJoins().isEmpty() ) {
+			return true;
+		}
+		return hasJoins( rootTableGroup.getTableGroupJoins() ) || hasJoins( rootTableGroup.getNestedTableGroupJoins() );
+	}
+
+	private static boolean hasJoins(List<TableGroupJoin> tableGroupJoins) {
+		for ( TableGroupJoin tableGroupJoin : tableGroupJoins ) {
+			final TableGroup joinedGroup = tableGroupJoin.getJoinedGroup();
+			if ( joinedGroup instanceof LazyTableGroup && ( (LazyTableGroup) joinedGroup ).getUnderlyingTableGroup() == null ) {
+				continue;
+			}
+			return true;
+		}
+		return false;
+	}
+
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// Insert-values statement
 
@@ -3097,6 +3115,32 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			implicitJoinChecker = BaseSqmToSqlAstConverter::verifyManipulationImplicitJoin;
 		}
 		prepareReusablePath( fromClauseIndex, sqmPath, implicitJoinChecker );
+
+		// Create the table group for every path that can potentially require one,
+		// as some paths require joining the target table i.e. inverse one-to-one
+		// Note that this will not necessarily create joins immediately, as table groups are lazy
+		if ( sqmPath instanceof SqmEntityValuedSimplePath<?>
+				|| sqmPath instanceof SqmEmbeddedValuedSimplePath<?>
+				|| sqmPath instanceof SqmAnyValuedSimplePath<?> ) {
+			final TableGroup existingTableGroup = fromClauseIndex.findTableGroup( sqmPath.getNavigablePath() );
+			if ( existingTableGroup == null ) {
+				final TableGroup createdTableGroup = createTableGroup(
+						fromClauseIndex.getTableGroup( sqmPath.getLhs().getNavigablePath() ),
+						sqmPath
+				);
+				if ( createdTableGroup != null ) {
+					if ( sqmPath instanceof SqmTreatedPath<?, ?> ) {
+						fromClauseIndex.register( sqmPath, createdTableGroup );
+					}
+					if ( sqmPath instanceof SqmFrom<?, ?> ) {
+						registerTreatUsage( (SqmFrom<?, ?>) sqmPath, createdTableGroup );
+					}
+				}
+			}
+			else if ( sqmPath instanceof SqmFrom<?, ?> ) {
+				registerTreatUsage( (SqmFrom<?, ?>) sqmPath, existingTableGroup );
+			}
+		}
 		return supplier.get();
 	}
 
@@ -3190,11 +3234,13 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		if ( tableGroup == null ) {
 			prepareReusablePath( path, () -> null );
 
+			if ( !( path instanceof SqmEntityValuedSimplePath<?>
+					|| path instanceof SqmEmbeddedValuedSimplePath<?>
+					|| path instanceof SqmAnyValuedSimplePath<?> ) ) {
+				// Since this is a selection, we must create a table group for the path as a DomainResult will be created
+				// But only create it for paths that are not handled by #prepareReusablePath anyway
 				final NavigablePath navigablePath;
-			if ( CollectionPart.Nature.fromNameExact( path.getNavigablePath().getLocalName() ) != null ) {
+				if ( path instanceof SqmTreatedRoot<?, ?> ) {
-				navigablePath = path.getLhs().getLhs().getNavigablePath();
-			}
-			else if ( path instanceof SqmTreatedRoot<?, ?> ) {
 					navigablePath = ( (SqmTreatedRoot<?, ?>) path ).getWrappedPath().getNavigablePath();
 				}
 				else {
@@ -3213,6 +3259,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 					}
 				}
 			}
+		}
 		else if ( path instanceof SqmFrom<?, ?> ) {
 			registerTreatUsage( (SqmFrom<?, ?>) path, tableGroup );
 		}
@@ -3288,7 +3335,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			}
 
 			fromClauseIndex.register( joinedPath, tableGroup );
-			registerPluralTableGroupParts( tableGroup );
+			registerPluralTableGroupParts( joinedPath.getNavigablePath(), tableGroup );
 		}
 		else {
 			tableGroup = null;
@@ -3322,17 +3369,25 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 	}
 
 	private void registerPluralTableGroupParts(TableGroup tableGroup) {
+		registerPluralTableGroupParts( null, tableGroup );
+	}
+
+	private void registerPluralTableGroupParts(NavigablePath navigablePath, TableGroup tableGroup) {
 		if ( tableGroup instanceof PluralTableGroup ) {
 			final PluralTableGroup pluralTableGroup = (PluralTableGroup) tableGroup;
 			if ( pluralTableGroup.getElementTableGroup() != null ) {
 				getFromClauseAccess().registerTableGroup(
-						pluralTableGroup.getElementTableGroup().getNavigablePath(),
+						navigablePath == null || navigablePath == tableGroup.getNavigablePath()
+								? pluralTableGroup.getElementTableGroup().getNavigablePath()
+								: navigablePath.append( CollectionPart.Nature.ELEMENT.getName() ),
 						pluralTableGroup.getElementTableGroup()
 				);
 			}
 			if ( pluralTableGroup.getIndexTableGroup() != null ) {
 				getFromClauseAccess().registerTableGroup(
-						pluralTableGroup.getIndexTableGroup().getNavigablePath(),
+						navigablePath == null || navigablePath == tableGroup.getNavigablePath()
+								? pluralTableGroup.getIndexTableGroup().getNavigablePath()
+								: navigablePath.append( CollectionPart.Nature.INDEX.getName() ),
 						pluralTableGroup.getIndexTableGroup()
 				);
 			}
@@ -3422,95 +3477,130 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 	}
 
 	private Expression visitTableGroup(TableGroup tableGroup, SqmFrom<?, ?> path) {
-		final ModelPartContainer modelPart;
+		final ModelPartContainer tableGroupModelPart = tableGroup.getModelPart();
-		final MappingModelExpressible<?> inferredValueMapping = getInferredValueMapping();
+		final ModelPart actualModelPart;
-		// For plain SqmFrom node uses, prefer the mapping type from the context if possible
+		final NavigablePath navigablePath;
-		if ( !( inferredValueMapping instanceof ModelPartContainer ) ) {
+		if ( tableGroupModelPart instanceof PluralAttributeMapping ) {
-			modelPart = tableGroup.getModelPart();
+			actualModelPart = ( (PluralAttributeMapping) tableGroupModelPart ).getElementDescriptor();
+			navigablePath = tableGroup.getNavigablePath().append( actualModelPart.getPartName() );
 		}
 		else {
-			modelPart = (ModelPartContainer) inferredValueMapping;
+			actualModelPart = tableGroupModelPart;
+			navigablePath = tableGroup.getNavigablePath();
 		}
+		final Expression result;
+		if ( actualModelPart instanceof EntityValuedModelPart ) {
+			final EntityValuedModelPart entityValuedModelPart = (EntityValuedModelPart) actualModelPart;
+			final EntityValuedModelPart inferredEntityMapping = (EntityValuedModelPart) getInferredValueMapping();
 			final ModelPart resultModelPart;
 			final ModelPart interpretationModelPart;
-		final TableGroup parentGroupToUse;
+			final TableGroup tableGroupToUse;
-		if ( modelPart instanceof ToOneAttributeMapping ) {
+			if ( inferredEntityMapping == null ) {
-			final ToOneAttributeMapping toOneAttributeMapping = (ToOneAttributeMapping) modelPart;
+				// When the inferred mapping is null, we try to resolve to the FK by default,
-			final ModelPart targetPart = toOneAttributeMapping.getForeignKeyDescriptor().getPart(
+				// which is fine because expansion to all target columns for select and group by clauses is handled below
-					toOneAttributeMapping.getSideNature().inverse()
+				if ( entityValuedModelPart instanceof EntityAssociationMapping && ( (EntityAssociationMapping) entityValuedModelPart ).getSideNature() != ForeignKeyDescriptor.Nature.TARGET ) {
+					// If the table group uses an association mapping that is not a one-to-many,
+					// we make use of the FK model part
+					final EntityAssociationMapping associationMapping = (EntityAssociationMapping) entityValuedModelPart;
+					final ModelPart targetPart = associationMapping.getForeignKeyDescriptor().getPart(
+							associationMapping.getSideNature()
 					);
-			if ( tableGroup.getModelPart().getPartMappingType() == modelPart.getPartMappingType() ) {
+					if ( entityValuedModelPart.getPartMappingType() == associationMapping.getPartMappingType() ) {
 						resultModelPart = targetPart;
 					}
 					else {
 						// If the table group is for a different mapping type i.e. an inheritance subtype,
 						// lookup the target part on that mapping type
-				resultModelPart = tableGroup.getModelPart().findSubPart( targetPart.getPartName(), null );
+						resultModelPart = entityValuedModelPart.findSubPart( targetPart.getPartName(), null );
-			}
-			interpretationModelPart = modelPart;
-			parentGroupToUse = null;
-		}
-		else if ( modelPart instanceof PluralAttributeMapping ) {
-			final PluralAttributeMapping pluralAttributeMapping = (PluralAttributeMapping) modelPart;
-			final CollectionPart elementDescriptor = pluralAttributeMapping.getElementDescriptor();
-			if ( elementDescriptor instanceof EntityCollectionPart ) {
-				// Usually, we need to resolve to the PK for visitTableGroup
-				final EntityCollectionPart collectionPart = (EntityCollectionPart) elementDescriptor;
-				final ModelPart collectionTargetPart = collectionPart.getForeignKeyDescriptor()
-						.getPart( collectionPart.getSideNature().inverse() );
-				final EntityIdentifierMapping identifierMapping = collectionPart.getEntityMappingType()
-						.getIdentifierMapping();
-				// If the FK points to the PK, we can use the FK part though, if this is not a root
-				if ( collectionTargetPart == identifierMapping && !( path instanceof SqmRoot<?> ) ) {
-					resultModelPart = collectionPart.getForeignKeyDescriptor().getPart( collectionPart.getSideNature() );
-				}
-				else {
-					resultModelPart = identifierMapping;
 					}
 				}
 				else {
-				resultModelPart = elementDescriptor;
+					// Otherwise, we use the identifier mapping of the target entity type
+					resultModelPart = entityValuedModelPart.getEntityMappingType().getIdentifierMapping();
 				}
-			interpretationModelPart = elementDescriptor;
+				interpretationModelPart = entityValuedModelPart;
-			parentGroupToUse = null;
+				tableGroupToUse = null;
 			}
-		else if ( modelPart instanceof EntityCollectionPart ) {
+			else if ( inferredEntityMapping instanceof ToOneAttributeMapping ) {
-			// Usually, we need to resolve to the PK for visitTableGroup
+				// If the inferred mapping is a to-one association,
-			final EntityCollectionPart collectionPart = (EntityCollectionPart) modelPart;
+				// we use the FK target part, which must be located on the entity mapping
-			final ModelPart collectionTargetPart = collectionPart.getForeignKeyDescriptor()
+				final ToOneAttributeMapping toOneAttributeMapping = (ToOneAttributeMapping) inferredEntityMapping;
-					.getPart( collectionPart.getSideNature().inverse() );
+				final ModelPart targetPart = toOneAttributeMapping.getForeignKeyDescriptor().getPart(
-			final EntityIdentifierMapping identifierMapping = collectionPart.getEntityMappingType()
+						toOneAttributeMapping.getSideNature().inverse()
-					.getIdentifierMapping();
+				);
-			// If the FK points to the PK, we can use the FK part though, if this is not a root
+				if ( entityValuedModelPart.getPartMappingType() == toOneAttributeMapping.getPartMappingType() ) {
-			if ( collectionTargetPart == identifierMapping && !( path instanceof SqmRoot<?> ) ) {
+					resultModelPart = targetPart;
-				resultModelPart = collectionPart.getForeignKeyDescriptor().getPart( collectionPart.getSideNature() );
 				}
 				else {
-				resultModelPart = identifierMapping;
+					// If the table group is for a different mapping type i.e. an inheritance subtype,
+					// lookup the target part on that mapping type
+					resultModelPart = entityValuedModelPart.findSubPart( targetPart.getPartName(), null );
 				}
-			interpretationModelPart = modelPart;
+				interpretationModelPart = toOneAttributeMapping;
-			parentGroupToUse = findTableGroup( tableGroup.getNavigablePath().getParent() );
+				tableGroupToUse = null;
 			}
-		else if ( modelPart instanceof EntityMappingType ) {
+			else if ( inferredEntityMapping instanceof EntityCollectionPart ) {
-			resultModelPart = ( (EntityMappingType) modelPart ).getIdentifierMapping();
+				// If the inferred mapping is a collection part, we try to make use of the FK again to avoid joins
-			interpretationModelPart = modelPart;
+				final EntityCollectionPart collectionPart = (EntityCollectionPart) inferredEntityMapping;
-			parentGroupToUse = null;
+				final TableGroup collectionTableGroup;
+				if ( tableGroup.getModelPart() instanceof CollectionPart ) {
+					collectionTableGroup = findTableGroup( tableGroup.getNavigablePath().getParent() );
 				}
 				else {
-			resultModelPart = modelPart;
+					collectionTableGroup = tableGroup;
-			interpretationModelPart = modelPart;
-			parentGroupToUse = null;
-		}
-		final NavigablePath navigablePath;
-		if ( interpretationModelPart == modelPart ) {
-			navigablePath = tableGroup.getNavigablePath();
-		}
-		else {
-			navigablePath = tableGroup.getNavigablePath().append( interpretationModelPart.getPartName() );
 				}
 
-		final Expression result;
+				if ( entityValuedModelPart == collectionPart ) {
-		if ( interpretationModelPart instanceof EntityValuedModelPart ) {
+					// When we compare the same collection parts, we can just use the FK part
+					resultModelPart = collectionPart.getForeignKeyDescriptor()
+							.getPart( collectionPart.getSideNature() );
+				}
+				else if ( entityValuedModelPart instanceof EntityAssociationMapping ) {
+					// If the table group model part is an association, we check if the FK targets are compatible
+					final EntityAssociationMapping tableGroupAssociation = (EntityAssociationMapping) entityValuedModelPart;
+					final ModelPart pathTargetPart = tableGroupAssociation.getForeignKeyDescriptor()
+							.getPart( tableGroupAssociation.getSideNature().inverse() );
+					final ModelPart inferredTargetPart = collectionPart.getForeignKeyDescriptor()
+							.getPart( collectionPart.getSideNature().inverse() );
+					// If the inferred association and table group association targets are the same,
+					// or the table group association refers to the primary key, we can safely use the FK part
+					if ( pathTargetPart == inferredTargetPart || tableGroupAssociation.isReferenceToPrimaryKey() ) {
+						resultModelPart = tableGroupAssociation.getForeignKeyDescriptor()
+								.getPart( tableGroupAssociation.getSideNature() );
+					}
+					else {
+						// Otherwise, we must force the use of the identifier mapping and possibly create a join,
+						// because comparing by primary key is the only sensible thing to do in this case.
+						// Note that EntityValuedPathInterpretation does the same
+						resultModelPart = collectionPart.getAssociatedEntityMappingType().getIdentifierMapping();
+					}
+				}
+				else {
+					// Since the table group model part is an EntityMappingType,
+					// we can render the FK target model part of the inferred collection part,
+					// which might be a UK, but usually a PK
+					assert entityValuedModelPart instanceof EntityMappingType;
+					if ( collectionPart.getCollectionDescriptor().isOneToMany() ) {
+						// When the inferred mapping is a one-to-many collection part,
+						// we will render the entity identifier mapping for that collection part,
+						// so we will have to do the same for the EntityMappingType side
+						resultModelPart = collectionPart.getAssociatedEntityMappingType().getIdentifierMapping();
+					}
+					else {
+						resultModelPart = collectionPart.getForeignKeyDescriptor()
+								.getPart( collectionPart.getSideNature().inverse() );
+					}
+				}
+				interpretationModelPart = inferredEntityMapping;
+				tableGroupToUse = collectionTableGroup;
+			}
+			else {
+				// Render the identifier mapping if the inferred mapping is an EntityMappingType
+				assert inferredEntityMapping instanceof EntityMappingType;
+				resultModelPart = ( (EntityMappingType) inferredEntityMapping ).getIdentifierMapping();
+				interpretationModelPart = inferredEntityMapping;
+				tableGroupToUse = null;
+			}
 			final boolean expandToAllColumns;
 			if ( currentClauseStack.getCurrent() == Clause.GROUP ) {
 				// When the table group is known to be fetched i.e. a fetch join
@@ -3523,28 +3613,29 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			}
 
 			final EntityValuedModelPart mapping = (EntityValuedModelPart) interpretationModelPart;
-			EntityMappingType mappingType;
+			final EntityMappingType treatedMapping;
 			if ( path instanceof SqmTreatedPath ) {
-				mappingType = creationContext.getSessionFactory()
+				treatedMapping = creationContext.getSessionFactory()
 						.getRuntimeMetamodels()
 						.getMappingMetamodel()
 						.findEntityDescriptor( ( (SqmTreatedPath<?,?>) path ).getTreatTarget().getHibernateEntityName() );
 			}
 			else {
-				mappingType = mapping.getEntityMappingType();
+				treatedMapping = mapping.getEntityMappingType();
 			}
 
 			result = EntityValuedPathInterpretation.from(
 					navigablePath,
-					parentGroupToUse == null ? tableGroup : parentGroupToUse,
+					tableGroupToUse == null ? tableGroup : tableGroupToUse,
 					expandToAllColumns ? null : resultModelPart,
+					true,
 					(EntityValuedModelPart) interpretationModelPart,
-					mappingType,
+					treatedMapping,
 					this
 			);
 		}
-		else if ( interpretationModelPart instanceof EmbeddableValuedModelPart ) {
+		else if ( actualModelPart instanceof EmbeddableValuedModelPart ) {
-			final EmbeddableValuedModelPart mapping = (EmbeddableValuedModelPart) resultModelPart;
+			final EmbeddableValuedModelPart mapping = (EmbeddableValuedModelPart) actualModelPart;
 			result = new EmbeddableValuedPathInterpretation<>(
 					mapping.toSqlExpression(
 							tableGroup,
@@ -3553,15 +3644,15 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 							getSqlAstCreationState()
 					),
 					navigablePath,
-					(EmbeddableValuedModelPart) interpretationModelPart,
+					mapping,
 					tableGroup
 			);
 		}
 		else {
-			assert interpretationModelPart instanceof BasicValuedModelPart;
+			assert actualModelPart instanceof BasicValuedModelPart;
-			final BasicValuedModelPart mapping = (BasicValuedModelPart) resultModelPart;
+			final BasicValuedModelPart mapping = (BasicValuedModelPart) actualModelPart;
 			final TableReference tableReference = tableGroup.resolveTableReference(
-					navigablePath.append( resultModelPart.getPartName() ),
+					navigablePath.append( actualModelPart.getPartName() ),
 					mapping.getContainingTableExpression()
 			);
 
@@ -3591,7 +3682,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			result = new BasicValuedPathInterpretation<>(
 					columnReference,
 					navigablePath,
-					(BasicValuedModelPart) interpretationModelPart,
+					mapping,
 					tableGroup
 			);
 		}
@@ -3599,7 +3690,6 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		return withTreatRestriction( result, path );
 	}
 
-
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// SqmPath
 
@@ -3756,15 +3846,18 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 
 	@Override
 	public Object visitFkExpression(SqmFkExpression<?> fkExpression) {
-		final EntityValuedPathInterpretation<?> toOneInterpretation = (EntityValuedPathInterpretation<?>) visitEntityValuedPath( fkExpression.getToOnePath() );
+		final SqmPath<?> lhs = fkExpression.getToOnePath().getLhs();
-		assert toOneInterpretation.getExpressionType() instanceof ToOneAttributeMapping;
+		prepareReusablePath( lhs, () -> null );
+		final TableGroup tableGroup = getFromClauseIndex().findTableGroup( lhs.getNavigablePath() );
+		final ModelPart subPart = tableGroup.getModelPart()
+				.findSubPart( fkExpression.getToOnePath().getModel().getPathName(), null );
+		assert subPart instanceof ToOneAttributeMapping;
 
-		final ToOneAttributeMapping toOneMapping = (ToOneAttributeMapping) toOneInterpretation.getExpressionType();
+		final ToOneAttributeMapping toOneMapping = (ToOneAttributeMapping) subPart;
 		final ForeignKeyDescriptor fkDescriptor = toOneMapping.getForeignKeyDescriptor();
-		final TableGroup tableGroup = toOneInterpretation.getTableGroup();
+		final TableReference tableReference = tableGroup.resolveTableReference( fkDescriptor.getTable( toOneMapping.getSideNature() ) );
-		final TableReference tableReference = tableGroup.resolveTableReference( fkDescriptor.getKeyTable() );
 
-		final ModelPart fkKeyPart = fkDescriptor.getKeyPart();
+		final ModelPart fkKeyPart = fkDescriptor.getPart( toOneMapping.getSideNature() );
 		if ( fkKeyPart instanceof BasicValuedModelPart ) {
 			final BasicValuedModelPart basicFkPart = (BasicValuedModelPart) fkKeyPart;
 
@@ -4686,9 +4779,16 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			final EntityValuedModelPart entityValuedModelPart = (EntityValuedModelPart) expressible;
 			final Object associationKey;
 			final ModelPart associationKeyPart;
-			if ( entityValuedModelPart instanceof Association ) {
+			if ( entityValuedModelPart instanceof EntityAssociationMapping ) {
-				final Association association = (Association) entityValuedModelPart;
+				final EntityAssociationMapping association = (EntityAssociationMapping) entityValuedModelPart;
 				final ForeignKeyDescriptor foreignKeyDescriptor = association.getForeignKeyDescriptor();
+				if ( association.getSideNature() == ForeignKeyDescriptor.Nature.TARGET ) {
+					// If the association is the target, we must use the identifier of the EntityMappingType
+					associationKey = association.getAssociatedEntityMappingType().getIdentifierMapping()
+							.getIdentifier( literal.getLiteralValue() );
+					associationKeyPart = association.getAssociatedEntityMappingType().getIdentifierMapping();
+				}
+				else {
 					associationKey = foreignKeyDescriptor.getAssociationKeyFromSide(
 							literal.getLiteralValue(),
 							association.getSideNature().inverse(),
@@ -4696,6 +4796,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 					);
 					associationKeyPart = foreignKeyDescriptor.getPart( association.getSideNature() );
 				}
+			}
 			else {
 				final EntityIdentifierMapping identifierMapping = entityValuedModelPart.getEntityMappingType()
 						.getIdentifierMapping();
@@ -4974,11 +5075,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		final MappingModelExpressible<?> inferredMapping = resolveInferredType();
 		if ( inferredMapping != null ) {
 			if ( inferredMapping instanceof PluralAttributeMapping ) {
-				final CollectionPart elementDescriptor = ( (PluralAttributeMapping) inferredMapping ).getElementDescriptor();
+				return ( (PluralAttributeMapping) inferredMapping ).getElementDescriptor();
-				if ( elementDescriptor instanceof EntityCollectionPart ) {
-					return ( (EntityCollectionPart) elementDescriptor ).getEntityMappingType();
-				}
-				return elementDescriptor;
 			}
 			else if ( !( inferredMapping instanceof JavaObjectType ) ) {
 				// Never report back the "object type" as inferred type and instead rely on the value type
@@ -5005,7 +5102,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		if ( paramType == null ) {
 			final MappingModelExpressible<?> inferredValueMapping = getInferredValueMapping();
 			if ( inferredValueMapping != null ) {
-				return inferredValueMapping;
+				return resolveInferredValueMappingForParameter( inferredValueMapping );
 			}
 			// Default to the Object type
 			return basicType( Object.class );
@@ -5019,7 +5116,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 				final JdbcMapping inferredJdbcMapping = inferredValueMapping.getJdbcMappings().get( 0 );
 				// If the bind type has a different JDBC type, we prefer that
 				if ( paramJdbcMapping.getJdbcType() == inferredJdbcMapping.getJdbcType() ) {
-					return inferredValueMapping;
+					return resolveInferredValueMappingForParameter( inferredValueMapping );
 				}
 			}
 			return paramModelType;
@@ -5030,7 +5127,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			// inferrable type and fallback to resolving the binding type
 			final MappingModelExpressible<?> inferredValueMapping = getInferredValueMapping();
 			if ( inferredValueMapping != null ) {
-				return inferredValueMapping;
+				return resolveInferredValueMappingForParameter( inferredValueMapping );
 			}
 		}
 		else if ( paramType instanceof EntityDomainType ) {
@@ -5039,7 +5136,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			// otherwise this would default to binding the PK which might be wrong
 			final MappingModelExpressible<?> inferredValueMapping = getInferredValueMapping();
 			if ( inferredValueMapping != null ) {
-				return inferredValueMapping;
+				return resolveInferredValueMappingForParameter( inferredValueMapping );
 			}
 		}
 
@@ -5060,7 +5157,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 				modelPart = getFromClauseAccess().getTableGroup( navigablePath ).getModelPart();
 			}
 			if ( modelPart instanceof PluralAttributeMapping ) {
-				return ( (PluralAttributeMapping) modelPart ).getElementDescriptor();
+				return resolveInferredValueMappingForParameter( ( (PluralAttributeMapping) modelPart ).getElementDescriptor() );
 			}
 			return modelPart;
 		}
@@ -5073,7 +5170,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			// Try to infer the value mapping since the other side apparently is a path source
 			final MappingModelExpressible<?> inferredValueMapping = getInferredValueMapping();
 			if ( inferredValueMapping != null ) {
-				return inferredValueMapping;
+				return resolveInferredValueMappingForParameter( inferredValueMapping );
 			}
 			throw new NotYetImplementedFor6Exception( "Support for embedded-valued parameters not yet implemented" );
 		}
@@ -5082,21 +5179,21 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 			// Try to infer the value mapping since the other side apparently is a path source
 			final MappingModelExpressible<?> inferredValueMapping = getInferredValueMapping();
 			if ( inferredValueMapping != null ) {
-				return inferredValueMapping;
+				return resolveInferredValueMappingForParameter( inferredValueMapping );
 			}
 
-			final BasicType basicTypeForJavaType = getTypeConfiguration().getBasicTypeForJavaType(
+			final BasicType<?> basicTypeForJavaType = getTypeConfiguration().getBasicTypeForJavaType(
 					paramSqmType.getExpressibleJavaType().getJavaTypeClass()
 			);
 
 			if ( basicTypeForJavaType == null ) {
 				if ( paramSqmType instanceof EntityDomainType ) {
-					return getIdType( (EntityDomainType) paramSqmType );
+					return resolveEntityPersister( (EntityDomainType<?>) paramSqmType );
 				}
 				else if ( paramSqmType instanceof SingularAttribute ) {
-					final Type type = ( (SingularAttribute) paramSqmType ).getType();
+					final Type<?> type = ( (SingularAttribute<?, ?>) paramSqmType ).getType();
 					if ( type instanceof EntityDomainType ) {
-						return getIdType( (EntityDomainType) type );
+						return resolveEntityPersister( (EntityDomainType<?>) type );
 					}
 				}
 			}
@@ -5107,13 +5204,16 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		throw new ConversionException( "Could not determine ValueMapping for SqmParameter: " + sqmParameter );
 	}
 
-	private BasicType getIdType(EntityDomainType entityDomainType) {
+	private MappingModelExpressible<?> resolveInferredValueMappingForParameter(MappingModelExpressible<?> inferredValueMapping) {
-		final SimpleDomainType idType = entityDomainType.getIdType();
+		if ( inferredValueMapping instanceof PluralAttributeMapping ) {
-		if ( idType != null ) {
+			// For parameters, we resolve to the element descriptor
-			return getTypeConfiguration().getBasicTypeForJavaType(
+			inferredValueMapping = ( (PluralAttributeMapping) inferredValueMapping ).getElementDescriptor();
-					idType.getExpressibleJavaType().getJavaTypeClass() );
 		}
-		return null;
+		if ( inferredValueMapping instanceof EntityCollectionPart ) {
+			// For parameters, we resolve to the entity mapping type to bind the primary key
+			return ( (EntityCollectionPart) inferredValueMapping ).getAssociatedEntityMappingType();
+		}
+		return inferredValueMapping;
 	}
 
 	private void resolveSqmParameter(
@@ -6293,16 +6393,7 @@ public abstract class BaseSqmToSqlAstConverter<T extends Statement> extends Base
 		final PluralAttributeMapping pluralAttributeMapping = (PluralAttributeMapping) determineValueMapping(
 				pluralPath );
 
-		if ( pluralAttributeMapping.getElementDescriptor() instanceof EntityCollectionPart ) {
-			inferrableTypeAccessStack.push(
-					() -> ( (EntityCollectionPart) pluralAttributeMapping.getElementDescriptor() ).getKeyTargetMatchPart() );
-		}
-		else if ( pluralAttributeMapping.getElementDescriptor() instanceof EmbeddedCollectionPart ) {
-			inferrableTypeAccessStack.push(pluralAttributeMapping::getElementDescriptor);
-		}
-		else {
 		inferrableTypeAccessStack.push( () -> pluralAttributeMapping );
-		}
 
 		final Expression lhs;
 		try {

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

@@ -13,6 +13,7 @@ import java.util.function.Consumer;
 
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.metamodel.mapping.BasicValuedModelPart;
+import org.hibernate.metamodel.mapping.EmbeddableValuedModelPart;
 import org.hibernate.metamodel.mapping.EntityAssociationMapping;
 import org.hibernate.metamodel.mapping.EntityDiscriminatorMapping;
 import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
@@ -21,16 +22,21 @@ import org.hibernate.metamodel.mapping.EntityValuedModelPart;
 import org.hibernate.metamodel.mapping.MappingModelExpressible;
 import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.SelectableConsumer;
+import org.hibernate.metamodel.mapping.internal.EntityCollectionPart;
 import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
 import org.hibernate.spi.NavigablePath;
 import org.hibernate.query.sqm.sql.SqmToSqlAstConverter;
 import org.hibernate.query.sqm.tree.domain.SqmEntityValuedSimplePath;
 import org.hibernate.sql.ast.SqlAstWalker;
+import org.hibernate.sql.ast.spi.FromClauseAccess;
+import org.hibernate.sql.ast.spi.SqlAstProcessingState;
+import org.hibernate.sql.ast.spi.SqlAstQueryPartProcessingState;
 import org.hibernate.sql.ast.spi.SqlExpressionResolver;
 import org.hibernate.sql.ast.tree.expression.ColumnReference;
 import org.hibernate.sql.ast.tree.expression.Expression;
 import org.hibernate.sql.ast.tree.expression.SqlTuple;
 import org.hibernate.sql.ast.tree.expression.SqlTupleContainer;
+import org.hibernate.sql.ast.tree.from.CorrelatedTableGroup;
 import org.hibernate.sql.ast.tree.from.TableGroup;
 import org.hibernate.sql.ast.tree.from.TableReference;
 import org.hibernate.sql.ast.tree.update.Assignable;
@@ -50,13 +56,8 @@ public class EntityValuedPathInterpretation<T> extends AbstractSqmPathInterpreta
 			SqmToSqlAstConverter sqlAstCreationState) {
 		final TableGroup tableGroup = sqlAstCreationState
 				.getFromClauseAccess()
-				.findTableGroup( sqmPath.getLhs().getNavigablePath() );
+				.findTableGroup( sqmPath.getNavigablePath() );
-		final EntityValuedModelPart pathMapping = (EntityValuedModelPart) sqlAstCreationState
+		final EntityValuedModelPart pathMapping = (EntityValuedModelPart) tableGroup.getModelPart();
-				.getFromClauseAccess()
-				.findTableGroup( sqmPath.getLhs().getNavigablePath() )
-				.getModelPart()
-				.findSubPart( sqmPath.getReferencedPathSource().getPathName(), null );
-		final EntityValuedModelPart mapping;
 		if ( inferredMapping instanceof EntityAssociationMapping ) {
 			final EntityAssociationMapping inferredAssociation = (EntityAssociationMapping) inferredMapping;
 			if ( pathMapping instanceof EntityAssociationMapping && inferredMapping != pathMapping ) {
@@ -67,54 +68,207 @@ public class EntityValuedPathInterpretation<T> extends AbstractSqmPathInterpreta
 						.getPart( pathAssociation.getSideNature().inverse() );
 				final ModelPart inferredTargetPart = inferredAssociation.getForeignKeyDescriptor()
 						.getPart( inferredAssociation.getSideNature().inverse() );
+
 				// If the inferred association and path association targets are the same, we can use the path mapping type
 				// which will render the FK of the path association
 				if ( pathTargetPart == inferredTargetPart ) {
-					mapping = pathMapping;
+					return from(
+							sqmPath.getNavigablePath(),
+							tableGroup,
+							pathMapping,
+							inferredMapping,
+							sqlAstCreationState
+					);
+				}
+				// When the inferred mapping and the path mapping differ,
+				// we always render the PK for the inferred and mapping path
+				else {
+					// If the path mapping refers to the primary key though,
+					// we can also render FK as that is equivalent
+					if ( pathAssociation.isReferenceToPrimaryKey() ) {
+						return from(
+								sqmPath.getNavigablePath(),
+								tableGroup,
+								pathMapping,
+								inferredMapping,
+								sqlAstCreationState
+						);
 					}
 					else {
-					// Otherwise, we need to use the entity mapping type to force rendering the PK
+						// but if the association FK does not point to the primary key,
-					// for e.g. `a.assoc1 = a.assoc2` when both associations have different target join columns
+						// we can't allow FK optimizations as that is problematic for self-referential associations,
-					mapping = pathMapping.getEntityMappingType();
+						// because then we would render the PK of the association owner instead of the target
+						return from(
+								sqmPath.getNavigablePath(),
+								tableGroup,
+								pathMapping.getEntityMappingType().getIdentifierMapping(),
+								false,
+								pathMapping,
+								pathMapping,
+								sqlAstCreationState
+						);
+					}
 				}
 			}
 			else {
 				// This is the case when the inferred mapping is an association, but the path mapping is not,
-				// or the path mapping and the inferred mapping are for the same association
+				// or the path mapping and the inferred mapping are for the same association,
-				mapping = (EntityValuedModelPart) inferredMapping;
+				// in which case we render this path like the inferred mapping
+				return from(
+						sqmPath.getNavigablePath(),
+						tableGroup,
+						(EntityValuedModelPart) inferredMapping,
+						inferredMapping,
+						sqlAstCreationState
+				);
 			}
 		}
 		else {
-			mapping = pathMapping;
+			// No inferred mapping available or it refers to an EntityMappingType
+			return from(
+					sqmPath.getNavigablePath(),
+					tableGroup,
+					pathMapping,
+					inferredMapping,
+					sqlAstCreationState
+			);
 		}
+	}
+
+	private static <T> EntityValuedPathInterpretation<T> from(
+			NavigablePath navigablePath,
+			TableGroup tableGroup,
+			EntityValuedModelPart mapping,
+			MappingModelExpressible<?> inferredMapping,
+			SqmToSqlAstConverter sqlAstCreationState) {
+		final boolean allowFkOptimization;
 		final ModelPart resultModelPart;
+		final TableGroup resultTableGroup;
+		// For association mappings where the FK optimization i.e. use of the parent table group is allowed,
+		// we try to make use of it and the FK model part if possible based on the inferred mapping
 		if ( mapping instanceof EntityAssociationMapping ) {
 			final EntityAssociationMapping associationMapping = (EntityAssociationMapping) mapping;
-			final ModelPart keyTargetMatchPart = associationMapping.getKeyTargetMatchPart();
+			final ModelPart associationKeyTargetMatchPart = associationMapping.getKeyTargetMatchPart();
-			if ( keyTargetMatchPart instanceof ToOneAttributeMapping ) {
+			final ModelPart keyTargetMatchPart;
-				resultModelPart = ( (ToOneAttributeMapping) keyTargetMatchPart ).getKeyTargetMatchPart();
+			if ( associationKeyTargetMatchPart instanceof ToOneAttributeMapping ) {
+				keyTargetMatchPart = ( (ToOneAttributeMapping) associationKeyTargetMatchPart ).getKeyTargetMatchPart();
+			}
+			else {
+				keyTargetMatchPart = associationKeyTargetMatchPart;
+			}
+
+			if ( associationMapping.isFkOptimizationAllowed() ) {
+				final boolean forceUsingForeignKeyAssociationSidePart;
+				// The following is an optimization for EntityCollectionPart path mappings with a join table.
+				// We can possibly avoid doing the join of the target table by using the parent table group
+				// and forcing to use the foreign key part of the association side
+				if ( inferredMapping != null && hasJoinTable( associationMapping ) ) {
+					// But we need to make sure the inferred mapping points to the same FK target
+					if ( inferredMapping instanceof EntityMappingType ) {
+						// If the inferred mapping is an EntityMappingType, meaning it's some sort of root path,
+						// we only have to make sure the FK target is part of the mapping type
+						forceUsingForeignKeyAssociationSidePart = ( (EntityMappingType) inferredMapping ).findSubPart( keyTargetMatchPart.getPartName() ) != null;
+					}
+					else {
+						// Otherwise, it must be an association mapping
+						assert inferredMapping instanceof EntityAssociationMapping;
+						// Comparing UK-based toOne with PK-based collection part
+						// In this case, we compare based on PK/FK, and can use the FK part if it points to the PK
+						forceUsingForeignKeyAssociationSidePart = ( (EntityAssociationMapping) inferredMapping ).getKeyTargetMatchPart() != keyTargetMatchPart
+								&& associationMapping.isReferenceToPrimaryKey();
+					}
+				}
+				else {
+					forceUsingForeignKeyAssociationSidePart = false;
+				}
+				if ( forceUsingForeignKeyAssociationSidePart ) {
+					if ( associationKeyTargetMatchPart instanceof ToOneAttributeMapping ) {
+						resultModelPart = keyTargetMatchPart;
+					}
+					else {
+						resultModelPart = associationMapping.getForeignKeyDescriptor()
+								.getPart( associationMapping.getSideNature() );
+					}
+					resultTableGroup = sqlAstCreationState.getFromClauseAccess()
+							.findTableGroup( tableGroup.getNavigablePath().getParent() );
+				}
+				else {
+					if ( isCorrelated( tableGroup, sqlAstCreationState ) ) {
+						// Access to the parent table group is forbidden for correlated table groups. For more details,
+						// see: `ToOneAttributeMapping.createRootTableGroupJoin`
+						// Due to that, we forcefully use the model part to which this association points to i.e. the target
+						resultModelPart = associationMapping.getForeignKeyDescriptor()
+								.getPart( associationMapping.getSideNature().inverse() );
 					}
 					else {
 						resultModelPart = keyTargetMatchPart;
 					}
+					resultTableGroup = tableGroup;
+				}
+				allowFkOptimization = true;
+			}
+			else if ( inferredMapping == null && hasNotFound( mapping ) ) {
+				// This is necessary to allow expression like `where root.notFoundAssociation is null`
+				// to render to `alias.not_found_fk is null`, but IMO this shouldn't be done
+				// todo: discuss removing this part and create a left joined table group instead?
+				resultModelPart = keyTargetMatchPart;
+				resultTableGroup = sqlAstCreationState.getFromClauseAccess()
+						.findTableGroup( tableGroup.getNavigablePath().getParent() );
+				allowFkOptimization = false;
 			}
 			else {
+				// If the mapping is an inverse association, use the PK and disallow FK optimizations
+				resultModelPart = ( (EntityAssociationMapping) mapping ).getAssociatedEntityMappingType().getIdentifierMapping();
+				resultTableGroup = tableGroup;
+				allowFkOptimization = false;
+			}
+		}
+		else {
+			// If the mapping is not an association, use the PK and disallow FK optimizations
 			resultModelPart = mapping.getEntityMappingType().getIdentifierMapping();
+			resultTableGroup = tableGroup;
+			allowFkOptimization = false;
 		}
 		return from(
-				sqmPath.getNavigablePath(),
+				navigablePath,
-				tableGroup,
+				resultTableGroup,
 				resultModelPart,
+				allowFkOptimization,
 				mapping,
 				mapping,
 				sqlAstCreationState
 		);
 	}
 
+	private static boolean isCorrelated(TableGroup tableGroup, SqmToSqlAstConverter sqlAstCreationState) {
+		final SqlAstProcessingState processingState = sqlAstCreationState.getCurrentProcessingState();
+		if ( !( processingState instanceof SqlAstQueryPartProcessingState )
+				|| ( (SqlAstQueryPartProcessingState) processingState ).getInflightQueryPart().isRoot() ) {
+			return false;
+		}
+		final FromClauseAccess fromClauseAccess = sqlAstCreationState.getFromClauseAccess();
+
+		TableGroup realParentTableGroup = fromClauseAccess.findTableGroup( tableGroup.getNavigablePath().getParent() );
+		while ( realParentTableGroup.getModelPart() instanceof EmbeddableValuedModelPart ) {
+			realParentTableGroup = fromClauseAccess.findTableGroup( realParentTableGroup.getNavigablePath().getParent() );
+		}
+		return realParentTableGroup instanceof CorrelatedTableGroup;
+	}
+
+	private static boolean hasNotFound(EntityValuedModelPart mapping) {
+		return mapping instanceof ToOneAttributeMapping && ( (ToOneAttributeMapping) mapping ).hasNotFoundAction();
+	}
+
+	private static boolean hasJoinTable(EntityAssociationMapping associationMapping) {
+		return associationMapping instanceof EntityCollectionPart
+				&& !( (EntityCollectionPart) associationMapping ).getCollectionDescriptor().isOneToMany();
+	}
+
 	public static <T> EntityValuedPathInterpretation<T> from(
 			NavigablePath navigablePath,
 			TableGroup tableGroup,
 			ModelPart resultModelPart,
+			boolean allowFkOptimization,
 			EntityValuedModelPart mapping,
 			EntityValuedModelPart treatedMapping,
 			SqmToSqlAstConverter sqlAstCreationState) {
@@ -163,7 +317,8 @@ public class EntityValuedPathInterpretation<T> extends AbstractSqmPathInterpreta
 				final BasicValuedModelPart basicValuedModelPart = (BasicValuedModelPart) resultModelPart;
 				final TableReference tableReference = tableGroup.resolveTableReference(
 						navigablePath,
-						basicValuedModelPart.getContainingTableExpression()
+						basicValuedModelPart.getContainingTableExpression(),
+						allowFkOptimization
 				);
 				sqlExpression = sqlExprResolver.resolveSqlExpression(
 						createColumnReferenceKey( tableReference, basicValuedModelPart.getSelectionExpression() ),
@@ -180,7 +335,8 @@ public class EntityValuedPathInterpretation<T> extends AbstractSqmPathInterpreta
 						(selectionIndex, selectableMapping) -> {
 							final TableReference tableReference = tableGroup.resolveTableReference(
 									navigablePath,
-									selectableMapping.getContainingTableExpression()
+									selectableMapping.getContainingTableExpression(),
+									allowFkOptimization
 							);
 							expressions.add(
 									sqlExprResolver.resolveSqlExpression(

hibernate-core/src/main/java/org/hibernate/spi/NavigablePath.java

@@ -235,7 +235,8 @@ public class NavigablePath implements DotIdentifierSequence, Serializable {
 
 		public String resolve() {
 			if ( buffer == null ) {
-				return null;
+				// Return an empty string instead of null in case the two navigable paths are equal
+				return matchedBase ? "" : null;
 			}
 			return buffer.toString();
 		}

hibernate-core/src/test/java/org/hibernate/orm/test/hql/CollectionMapWithComponentValueTest.java

@@ -202,6 +202,17 @@ public class CollectionMapWithComponentValueTest extends BaseCoreFunctionalTestC
 		} );
 	}
 
+	@Test
+	@TestForIssue(jiraKey = "HHH-11433")
+	public void testJoinMapValue() {
+		doInHibernate( this::sessionFactory, s -> {
+			// Assert that a left join is used for joining the map key entity table
+			List keyValues= s.createQuery( "select v from BaseTestEntity bte left join bte.entities te left join te.values v" ).list();
+			System.out.println( keyValues );
+			assertEquals( 2, keyValues.size() );
+		} );
+	}
+
 	@Test
 	@TestForIssue(jiraKey = "HHH-11433")
 	public void testJoinMapKey() {

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

@@ -0,0 +1,406 @@
+package org.hibernate.orm.test.query;
+
+import java.util.Set;
+
+import org.hibernate.testing.TestForIssue;
+import org.hibernate.testing.jdbc.SQLStatementInspector;
+import org.hibernate.testing.orm.junit.DomainModel;
+import org.hibernate.testing.orm.junit.SessionFactory;
+import org.hibernate.testing.orm.junit.SessionFactoryScope;
+import org.junit.jupiter.api.Test;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.FetchType;
+import jakarta.persistence.Id;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.JoinTable;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.OneToMany;
+import jakarta.persistence.Table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@DomainModel(
+		annotatedClasses = { CompareEntityValuedPathsTest.Person.class }
+)
+@SessionFactory(statementInspectorClass = SQLStatementInspector.class)
+@TestForIssue(jiraKey = "HHH-15349")
+public class CompareEntityValuedPathsTest {
+
+	@Test
+	public void testCompareOneToManyUK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p in (select c from p.childrenUk c)" ).list();
+
+					// Ensure that there are no joins and we compare by UK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.uk in(" +
+									"select c1_0.child_uk " +
+									"from children_uks c1_0 " +
+									"where p1_0.uk=c1_0.owner_uk" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareOneToManyPK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p in (select c from p.children c)" ).list();
+
+					// Ensure that there are no joins and we compare by PK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.id in(" +
+									"select c1_0.children_id " +
+									"from PERSON_TABLE_PERSON_TABLE c1_0 " +
+									"where p1_0.id=c1_0.Person_id" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testManyToOneIsNull(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p.parent is null" ).list();
+
+					// Comparing a PK-ManyToOne against a UK-CollectionPart will force comparing by PK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.parent_id is null",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testComparePKWithOneToManyUK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p.parent in (select c from p.childrenUk c)" ).list();
+
+					// Comparing a PK-ManyToOne against a UK-CollectionPart will force comparing by PK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.parent_id in(" +
+									"select c1_1.id " +
+									"from children_uks c1_0 " +
+									"join PERSON_TABLE c1_1 on c1_1.uk=c1_0.child_uk " +
+									"where p1_0.uk=c1_0.owner_uk" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareUKWithOneToManyPK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p.parentUk in (select c from p.children c)" ).list();
+
+					// Comparing a UK-ManyToOne against a PK-CollectionPart will force comparing by PK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join PERSON_TABLE p2_0 on p2_0.uk=p1_0.parent_uk " +
+									"where p2_0.id in(" +
+									"select c1_0.children_id " +
+									"from PERSON_TABLE_PERSON_TABLE c1_0 " +
+									"where p1_0.id=c1_0.Person_id" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testComparePKWithMappedByOneToManyPK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p in (select c from p.employees c)" ).list();
+
+					// Comparing a root against a PK-one-to-many CollectionPart allows comparing by FK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.id in(" +
+									"select e1_0.id " +
+									"from PERSON_TABLE e1_0 " +
+									"where p1_0.id=e1_0.supervisor_id" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testComparePKWithMappedByOneToManyUK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p in (select c from p.employeesUk c)" ).list();
+
+					// Comparing a root against a UK-one-to-many CollectionPart forces comparing by PK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.id in(" +
+									"select e1_0.id " +
+									"from PERSON_TABLE e1_0 " +
+									"where p1_0.uk=e1_0.supervisor_uk" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testOneToManyUKIsNotNull(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p join p.childrenUk c where c is not null" ).list();
+
+					// Assert that we don't join the childrenUk target table, only the join table
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join children_uks c1_0 on p1_0.uk=c1_0.owner_uk " +
+									"where c1_0.child_uk is not null",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testOneToManyPKIsNotNull(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p join p.children c where c is not null" ).list();
+
+					// Assert that we don't join the children target table, only the join table
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join PERSON_TABLE_PERSON_TABLE c1_0 on p1_0.id=c1_0.Person_id " +
+									"where c1_0.children_id is not null",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareOneToManyUKWithOneToManyPK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p join p.childrenUk c join p.children c2 where c = c2" ).list();
+
+					// Assert that we don't join the childrenUk target table, only the join table
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join (children_uks c1_0 join PERSON_TABLE c1_1 on c1_1.uk=c1_0.child_uk) on p1_0.uk=c1_0.owner_uk " +
+									"join PERSON_TABLE_PERSON_TABLE c2_0 on p1_0.id=c2_0.Person_id " +
+									"where c1_1.id=c2_0.children_id",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareOneToManyPKWithOneToManyUK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p join p.children c join p.childrenUk c2 where c = c2" ).list();
+
+					// Assert that we don't join the children target table, only the join table
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join PERSON_TABLE_PERSON_TABLE c1_0 on p1_0.id=c1_0.Person_id " +
+									"join (children_uks c2_0 join PERSON_TABLE c2_1 on c2_1.uk=c2_0.child_uk) on p1_0.uk=c2_0.owner_uk " +
+									"where c1_0.children_id=c2_1.id",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareOneToManyUKWithSubqueryOneToManyPK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p join p.childrenUk c where c in (select c2 from p.children c2)" ).list();
+
+					// Assert that we don't join the childrenUk target table, only the join table
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join (children_uks c1_0 join PERSON_TABLE c1_1 on c1_1.uk=c1_0.child_uk) on p1_0.uk=c1_0.owner_uk " +
+									"where c1_1.id in(select c2_0.children_id from PERSON_TABLE_PERSON_TABLE c2_0 where p1_0.id=c2_0.Person_id)",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareOneToManyPKWithSubqueryOneToManyUK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p join p.children c where c in (select c2 from p.childrenUk c2)" ).list();
+
+					// Assert that we don't join the children target table, only the join table
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"join PERSON_TABLE_PERSON_TABLE c1_0 on p1_0.id=c1_0.Person_id " +
+									"where c1_0.children_id in(select c2_1.id from children_uks c2_0 join PERSON_TABLE c2_1 on c2_1.uk=c2_0.child_uk where p1_0.uk=c2_0.owner_uk)",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareManyToOneUK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p in (select p2.parentUk from Person p2)" ).list();
+
+					// Ensure that there are no joins and we compare by UK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.uk in(" +
+									"select p2_0.parent_uk " +
+									"from PERSON_TABLE p2_0" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Test
+	public void testCompareManyToOnePK(SessionFactoryScope scope) {
+		final SQLStatementInspector statementInspector = (SQLStatementInspector) scope.getStatementInspector();
+		scope.inTransaction(
+				session -> {
+					statementInspector.clear();
+					session.createQuery( "select 1 from Person p where p in (select p2.parent from Person p2)" ).list();
+
+					// Ensure that there are no joins and we compare by PK
+					assertEquals(
+							"select " +
+									"1 " +
+									"from PERSON_TABLE p1_0 " +
+									"where p1_0.id in(" +
+									"select p2_0.parent_id " +
+									"from PERSON_TABLE p2_0" +
+									")",
+							statementInspector.getSqlQueries().get( 0 )
+					);
+				}
+		);
+	}
+
+	@Entity(name = "Person")
+	@Table(name = "PERSON_TABLE")
+	public static class Person {
+		@Id
+		private Long id;
+
+		@Column(unique = true)
+		private String uk;
+		private String name;
+		@ManyToOne(fetch = FetchType.LAZY)
+		private Person parent;
+		@ManyToOne(fetch = FetchType.LAZY)
+		@JoinColumn(name = "parent_uk", referencedColumnName = "uk")
+		private Person parentUk;
+		@ManyToOne(fetch = FetchType.LAZY)
+		private Person supervisor;
+		@ManyToOne(fetch = FetchType.LAZY)
+		@JoinColumn(name = "supervisor_uk", referencedColumnName = "uk")
+		private Person supervisorUk;
+		@OneToMany
+		private Set<Person> children;
+		@OneToMany
+		@JoinTable(name = "children_uks",
+				joinColumns = @JoinColumn(name = "owner_uk", referencedColumnName = "uk"),
+				inverseJoinColumns = @JoinColumn(name = "child_uk", referencedColumnName = "uk")
+		)
+		private Set<Person> childrenUk;
+		@OneToMany(mappedBy = "supervisor")
+		private Set<Person> employees;
+		@OneToMany(mappedBy = "supervisorUk")
+		private Set<Person> employeesUk;
+
+		public Person() {
+		}
+	}
+}

hibernate-core/src/test/java/org/hibernate/orm/test/query/hql/EntityJoinTest.java

@@ -13,6 +13,7 @@ import jakarta.persistence.Id;
 import jakarta.persistence.JoinColumn;
 import jakarta.persistence.ManyToOne;
 import jakarta.persistence.Table;
+import org.hamcrest.CoreMatchers;
 
 import org.hibernate.annotations.NaturalId;
 import org.hibernate.dialect.SybaseDialect;
@@ -29,6 +30,7 @@ import org.hibernate.query.sqm.sql.SqmTranslator;
 import org.hibernate.query.sqm.sql.SqmTranslatorFactory;
 import org.hibernate.query.sqm.tree.SqmStatement;
 import org.hibernate.query.sqm.tree.select.SqmSelectStatement;
+import org.hibernate.sql.ast.tree.from.LazyTableGroup;
 import org.hibernate.sql.ast.tree.from.TableGroup;
 import org.hibernate.sql.ast.tree.from.TableGroupJoin;
 import org.hibernate.sql.ast.tree.select.SelectStatement;
@@ -43,6 +45,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import static org.hamcrest.core.Is.is;
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 
@@ -202,9 +205,14 @@ public class EntityJoinTest {
                     assertThat( roots.size(), is( 1 ) );
 
                     final TableGroup rootTableGroup = roots.get( 0 );
-                    assertThat( rootTableGroup.getTableGroupJoins().size(), is( 1 ) );
+                    assertThat( rootTableGroup.getTableGroupJoins().size(), is( 2 ) );
 
-                    final TableGroupJoin tableGroupJoin = rootTableGroup.getTableGroupJoins().get( 0 );
+                    // The first table group is an uninitialized lazy table group for the path in the on clause
+                    final TableGroupJoin firstTableGroupJoin = rootTableGroup.getTableGroupJoins().get( 0 );
+                    assertThat( firstTableGroupJoin.getJoinedGroup(), instanceOf( LazyTableGroup.class ) );
+                    assertThat( ((LazyTableGroup) firstTableGroupJoin.getJoinedGroup()).getUnderlyingTableGroup(), is( CoreMatchers.nullValue() ) );
+
+                    final TableGroupJoin tableGroupJoin = rootTableGroup.getTableGroupJoins().get( 1 );
                     assertThat( tableGroupJoin.getJoinedGroup().getModelPart(), is( customerEntityDescriptor ) );
                 }
         );