Fix collection table cleanup issues for all strategies

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

@@ -582,7 +582,7 @@ public class EmbeddableMappingType implements ManagedMappingType, SelectableMapp
 			BiConsumer<SqlSelection, JdbcMapping> selectionConsumer) {
 		visitAttributeMappings(
 				attributeMapping ->
-						ManagedMappingType.super.applySqlSelections(
+						attributeMapping.applySqlSelections(
 								navigablePath,
 								tableGroup,
 								creationState,
@@ -742,7 +742,7 @@ public class EmbeddableMappingType implements ManagedMappingType, SelectableMapp
 	}
 
 	@Override
-	public void visitAttributeMappings(Consumer<AttributeMapping> action) {
+	public void visitAttributeMappings(Consumer<? super AttributeMapping> action) {
 		attributeMappings.forEach( action );
 	}
 

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

@@ -209,7 +209,7 @@ public interface EntityMappingType extends ManagedMappingType, EntityValuedModel
 
 
 	@Override
-	default void visitAttributeMappings(Consumer<AttributeMapping> action) {
+	default void visitAttributeMappings(Consumer<? super AttributeMapping> action) {
 		getAttributeMappings().forEach( action );
 	}
 

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

@@ -47,7 +47,7 @@ public interface ManagedMappingType extends MappingType, FetchableContainer {
 	/**
 	 * Visit attributes defined on this class and any supers
 	 */
-	void visitAttributeMappings(Consumer<AttributeMapping> action);
+	void visitAttributeMappings(Consumer<? super AttributeMapping> action);
 
 	/**
 	 * Visit attributes defined on this class and any supers

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

@@ -14,6 +14,7 @@ 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.from.TableGroup;
 
 import static org.hibernate.sql.ast.spi.SqlExpressionResolver.createColumnReferenceKey;
 
@@ -25,6 +26,14 @@ public class MappingModelHelper {
 			ModelPart modelPart,
 			SqlExpressionResolver sqlExpressionResolver,
 			SessionFactoryImplementor sessionFactory) {
+		return buildColumnReferenceExpression( null, modelPart, sqlExpressionResolver, sessionFactory );
+	}
+
+	public static Expression buildColumnReferenceExpression(
+			TableGroup tableGroup,
+			ModelPart modelPart,
+			SqlExpressionResolver sqlExpressionResolver,
+			SessionFactoryImplementor sessionFactory) {
 		final int jdbcTypeCount = modelPart.getJdbcTypeCount();
 
 		if ( modelPart instanceof EmbeddableValuedModelPart ) {
@@ -32,9 +41,16 @@ public class MappingModelHelper {
 			modelPart.forEachSelectable(
 					(columnIndex, selection) -> {
 						final ColumnReference colRef;
+						final String qualifier;
+						if ( tableGroup == null ) {
+							qualifier = selection.getContainingTableExpression();
+						}
+						else {
+							qualifier = tableGroup.getTableReference( selection.getContainingTableExpression() ).getIdentificationVariable();
+						}
 						if ( sqlExpressionResolver == null ) {
 							colRef = new ColumnReference(
-									selection.getContainingTableExpression(),
+									qualifier,
 									selection,
 									sessionFactory
 							);
@@ -43,7 +59,7 @@ public class MappingModelHelper {
 							colRef = (ColumnReference) sqlExpressionResolver.resolveSqlExpression(
 									createColumnReferenceKey( selection.getContainingTableExpression(), selection.getSelectionExpression() ),
 									sqlAstProcessingState -> new ColumnReference(
-											selection.getContainingTableExpression(),
+											qualifier,
 											selection,
 											sessionFactory
 									)
@@ -57,9 +73,16 @@ public class MappingModelHelper {
 		else {
 			assert modelPart instanceof BasicValuedModelPart;
 			final BasicValuedModelPart basicPart = (BasicValuedModelPart) modelPart;
+			final String qualifier;
+			if ( tableGroup == null ) {
+				qualifier = basicPart.getContainingTableExpression();
+			}
+			else {
+				qualifier = tableGroup.getTableReference( basicPart.getContainingTableExpression() ).getIdentificationVariable();
+			}
 			if ( sqlExpressionResolver == null ) {
 				return new ColumnReference(
-						basicPart.getContainingTableExpression(),
+						qualifier,
 						basicPart,
 						sessionFactory
 				);
@@ -68,7 +91,7 @@ public class MappingModelHelper {
 				return sqlExpressionResolver.resolveSqlExpression(
 						createColumnReferenceKey( basicPart.getContainingTableExpression(), basicPart.getSelectionExpression() ),
 						sqlAstProcessingState -> new ColumnReference(
-								basicPart.getContainingTableExpression(),
+								qualifier,
 								basicPart,
 								sessionFactory
 						)

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

@@ -7,6 +7,7 @@
 package org.hibernate.metamodel.mapping.internal;
 
 import java.util.Locale;
+import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 
 import org.hibernate.engine.FetchStyle;
@@ -212,6 +213,18 @@ public class BasicEntityIdentifierMappingImpl implements BasicEntityIdentifierMa
 		resolveSqlSelection( navigablePath, tableGroup, true, creationState );
 	}
 
+	@Override
+	public void applySqlSelections(
+			NavigablePath navigablePath,
+			TableGroup tableGroup,
+			DomainResultCreationState creationState,
+			BiConsumer<SqlSelection, JdbcMapping> selectionConsumer) {
+		selectionConsumer.accept(
+				resolveSqlSelection( navigablePath, tableGroup, true, creationState ),
+				getJdbcMapping()
+		);
+	}
+
 	private SqlSelection resolveSqlSelection(
 			NavigablePath navigablePath,
 			TableGroup tableGroup,

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

@@ -8,6 +8,7 @@ package org.hibernate.metamodel.mapping.internal;
 
 import java.io.Serializable;
 import java.util.List;
+import java.util.function.BiConsumer;
 
 import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.engine.spi.PersistenceContext;
@@ -19,9 +20,14 @@ import org.hibernate.mapping.Property;
 import org.hibernate.metamodel.internal.AbstractCompositeIdentifierMapping;
 import org.hibernate.metamodel.mapping.EmbeddableMappingType;
 import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.JdbcMapping;
 import org.hibernate.metamodel.mapping.SingularAttributeMapping;
 import org.hibernate.metamodel.mapping.StateArrayContributorMetadataAccess;
 import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.query.NavigablePath;
+import org.hibernate.sql.ast.spi.SqlSelection;
+import org.hibernate.sql.ast.tree.from.TableGroup;
+import org.hibernate.sql.results.graph.DomainResultCreationState;
 import org.hibernate.type.ComponentType;
 
 /**
@@ -133,6 +139,32 @@ public class NonAggregatedIdentifierMappingImpl extends AbstractCompositeIdentif
 		}
 	}
 
+	@Override
+	public void applySqlSelections(
+			NavigablePath navigablePath,
+			TableGroup tableGroup,
+			DomainResultCreationState creationState) {
+		for ( int i = 0; i < idAttributeMappings.size(); i++ ) {
+			idAttributeMappings.get( i ).applySqlSelections( navigablePath, tableGroup, creationState );
+		}
+	}
+
+	@Override
+	public void applySqlSelections(
+			NavigablePath navigablePath,
+			TableGroup tableGroup,
+			DomainResultCreationState creationState,
+			BiConsumer<SqlSelection, JdbcMapping> selectionConsumer) {
+		for ( int i = 0; i < idAttributeMappings.size(); i++ ) {
+			idAttributeMappings.get( i ).applySqlSelections(
+					navigablePath,
+					tableGroup,
+					creationState,
+					selectionConsumer
+			);
+		}
+	}
+
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// EmbeddableValuedFetchable
 

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

@@ -10,6 +10,7 @@ import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
+import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.hibernate.LockMode;
@@ -50,6 +51,7 @@ import org.hibernate.sql.ast.spi.SqlAliasStemHelper;
 import org.hibernate.sql.ast.spi.SqlAstCreationContext;
 import org.hibernate.sql.ast.spi.SqlAstCreationState;
 import org.hibernate.sql.ast.spi.SqlExpressionResolver;
+import org.hibernate.sql.ast.spi.SqlSelection;
 import org.hibernate.sql.ast.tree.from.LazyTableGroup;
 import org.hibernate.sql.ast.tree.from.StandardTableGroup;
 import org.hibernate.sql.ast.tree.from.TableGroup;
@@ -1157,6 +1159,32 @@ public class ToOneAttributeMapping
 		}
 	}
 
+	@Override
+	public void applySqlSelections(
+			NavigablePath navigablePath,
+			TableGroup tableGroup,
+			DomainResultCreationState creationState) {
+		if ( sideNature == ForeignKeyDescriptor.Nature.KEY ) {
+			foreignKeyDescriptor.getKeyPart().applySqlSelections( navigablePath, tableGroup, creationState );
+		}
+	}
+
+	@Override
+	public void applySqlSelections(
+			NavigablePath navigablePath,
+			TableGroup tableGroup,
+			DomainResultCreationState creationState,
+			BiConsumer<SqlSelection, JdbcMapping> selectionConsumer) {
+		if ( sideNature == ForeignKeyDescriptor.Nature.KEY ) {
+			foreignKeyDescriptor.getKeyPart().applySqlSelections(
+					navigablePath,
+					tableGroup,
+					creationState,
+					selectionConsumer
+			);
+		}
+	}
+
 	@Override
 	public int getJdbcTypeCount() {
 		return foreignKeyDescriptor.getJdbcTypeCount();

hibernate-core/src/main/java/org/hibernate/persister/entity/AbstractEntityPersister.java

@@ -5452,7 +5452,7 @@ public abstract class AbstractEntityPersister
 	protected ReflectionOptimizer.AccessOptimizer accessOptimizer;
 
 	@Override
-	public void visitAttributeMappings(Consumer<AttributeMapping> action) {
+	public void visitAttributeMappings(Consumer<? super AttributeMapping> action) {
 		attributeMappings.forEach( action );
 	}
 
@@ -5540,7 +5540,6 @@ public abstract class AbstractEntityPersister
 				"Entity(" + getEntityName() + ") `staticFetchableList` generator",
 				() -> {
 					staticFetchableList = new ArrayList<>( attributeMappings.size() );
-					visitAttributeMappings( attributeMapping -> staticFetchableList.add( attributeMapping ) );
 					visitSubTypeAttributeMappings( attributeMapping -> staticFetchableList.add( attributeMapping ) );
 					return true;
 				}
@@ -6352,11 +6351,12 @@ public abstract class AbstractEntityPersister
 			return;
 		}
 
-		attributeMappings.forEach( fetchableConsumer );
-
 		if ( treatTargetType.isTypeOrSuperType( this ) ) {
 			visitSubTypeAttributeMappings( fetchableConsumer );
 		}
+		else {
+			attributeMappings.forEach( fetchableConsumer );
+		}
 	}
 
 	protected List<Fetchable> getStaticFetchableList() {
@@ -6389,11 +6389,11 @@ public abstract class AbstractEntityPersister
 
 	@Override
 	public void visitSubTypeAttributeMappings(Consumer<? super AttributeMapping> action) {
+		visitAttributeMappings( action );
 		if ( subclassMappingTypes != null ) {
 			subclassMappingTypes.forEach(
 					(s, subType) -> {
 						subType.visitDeclaredAttributeMappings( action );
-						subType.visitSubTypeAttributeMappings( action );
 					}
 			);
 		}

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

@@ -137,26 +137,28 @@ public class SimpleDeleteQueryPlan implements NonSelectQueryPlan {
 
 					final ForeignKeyDescriptor fkDescriptor = attributeMapping.getKeyDescriptor();
 					final Expression fkColumnExpression = MappingModelHelper.buildColumnReferenceExpression(
-							fkDescriptor,
+							fkDescriptor.getKeyPart(),
 							null,
 							factory
 					);
 
 					final QuerySpec matchingIdSubQuery = new QuerySpec( false );
 
+					final MutatingTableReferenceGroupWrapper tableGroup = new MutatingTableReferenceGroupWrapper(
+							new NavigablePath( attributeMapping.getRootPathName() ),
+							attributeMapping,
+							sqmInterpretation.getSqlAst().getTargetTable()
+					);
 					final Expression fkTargetColumnExpression = MappingModelHelper.buildColumnReferenceExpression(
-							fkDescriptor,
+							tableGroup,
+							fkDescriptor.getTargetPart(),
 							sqmInterpretation.getSqlExpressionResolver(),
 							factory
 					);
 					matchingIdSubQuery.getSelectClause().addSqlSelection( new SqlSelectionImpl( 1, 0, fkTargetColumnExpression ) );
 
 					matchingIdSubQuery.getFromClause().addRoot(
-							new MutatingTableReferenceGroupWrapper(
+							tableGroup
-									new NavigablePath( attributeMapping.getRootPathName() ),
-									attributeMapping,
-									sqmInterpretation.getSqlAst().getTargetTable()
-							)
 					);
 
 					matchingIdSubQuery.applyPredicate( sqmInterpretation.getSqlAst().getRestriction() );

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

@@ -18,13 +18,16 @@ import org.hibernate.engine.jdbc.env.spi.JdbcEnvironment;
 import org.hibernate.engine.jdbc.spi.JdbcServices;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.internal.FilterHelper;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.metamodel.model.domain.EntityDomainType;
 import org.hibernate.persister.entity.Joinable;
 import org.hibernate.query.spi.DomainQueryExecutionContext;
 import org.hibernate.query.sqm.internal.DomainParameterXref;
 import org.hibernate.query.sqm.internal.SqmJdbcExecutionContextAdapter;
 import org.hibernate.query.sqm.internal.SqmUtil;
+import org.hibernate.query.sqm.sql.internal.SqlAstQueryPartProcessingStateImpl;
 import org.hibernate.query.sqm.tree.SqmDeleteOrUpdateStatement;
 import org.hibernate.query.sqm.tree.expression.SqmParameter;
 import org.hibernate.sql.ast.SqlAstJoinType;
@@ -89,34 +92,29 @@ public class MatchingIdSelectionHelper {
 		idSelectionQuery.getFromClause().addRoot( mutatingTableGroup );
 
 		final List<DomainResult<?>> domainResults = new ArrayList<>();
-
+		sqmConverter.getProcessingStateStack().push(
-		targetEntityDescriptor.getIdentifierMapping().forEachSelectable(
+				new SqlAstQueryPartProcessingStateImpl(
-				(position, selection) -> {
+						idSelectionQuery,
-					final TableReference tableReference = mutatingTableGroup.resolveTableReference(
+						sqmConverter.getCurrentProcessingState(),
-							mutatingTableGroup.getNavigablePath(),
+						sqmConverter.getSqlAstCreationState(),
-							selection.getContainingTableExpression()
+						sqmConverter.getCurrentClauseStack()::getCurrent
-					);
+				)
-					final Expression expression = sqmConverter.getSqlExpressionResolver().resolveSqlExpression(
+		);
-							SqlExpressionResolver.createColumnReferenceKey( tableReference, selection.getSelectionExpression() ),
+		targetEntityDescriptor.getIdentifierMapping().applySqlSelections(
-							sqlAstProcessingState -> new ColumnReference(
+				mutatingTableGroup.getNavigablePath(),
-									tableReference,
+				mutatingTableGroup,
-									selection,
+				sqmConverter,
-									sessionFactory
+				(selection, jdbcMapping) -> {
+					domainResults.add(
+							new BasicResult<>(
+									selection.getValuesArrayPosition(),
+									null,
+									jdbcMapping.getJavaTypeDescriptor()
 							)
 					);
-					idSelectionQuery.getSelectClause().addSqlSelection(
-							new SqlSelectionImpl(
-									position,
-									position + 1,
-									expression
-							)
-					);
-
-					//noinspection unchecked
-					domainResults.add( new BasicResult( position, null, selection.getJdbcMapping().getJavaTypeDescriptor() ) );
-
 				}
 		);
+		sqmConverter.getProcessingStateStack().pop();
 
 		final FilterPredicate filterPredicate = FilterHelper.createFilterPredicate(
 				executionContext.getSession().getLoadQueryInfluencers(),
@@ -237,6 +235,46 @@ public class MatchingIdSelectionHelper {
 				factory
 		);
 
+		sqmConverter.getProcessingStateStack().push(
+				new SqlAstQueryPartProcessingStateImpl(
+						matchingIdSelection.getQuerySpec(),
+						sqmConverter.getCurrentProcessingState(),
+						sqmConverter.getSqlAstCreationState(),
+						sqmConverter.getCurrentClauseStack()::getCurrent
+				)
+		);
+		entityDescriptor.visitSubTypeAttributeMappings(
+				attribute -> {
+					if ( attribute instanceof PluralAttributeMapping ) {
+						final PluralAttributeMapping pluralAttribute = (PluralAttributeMapping) attribute;
+
+						if ( pluralAttribute.getSeparateCollectionTable() != null ) {
+							// Ensure that the FK target columns are available
+							final boolean useFkTarget = !( pluralAttribute.getKeyDescriptor()
+									.getTargetPart() instanceof EntityIdentifierMapping );
+							if ( useFkTarget ) {
+								final TableGroup mutatingTableGroup = sqmConverter.getMutatingTableGroup();
+								pluralAttribute.getKeyDescriptor().getTargetPart().applySqlSelections(
+										mutatingTableGroup.getNavigablePath(),
+										mutatingTableGroup,
+										sqmConverter,
+										(selection, jdbcMapping) -> {
+											matchingIdSelection.getDomainResultDescriptors().add(
+													new BasicResult<>(
+															selection.getValuesArrayPosition(),
+															null,
+															jdbcMapping.getJavaTypeDescriptor()
+													)
+											);
+										}
+								);
+							}
+						}
+					}
+				}
+		);
+		sqmConverter.getProcessingStateStack().pop();
+
 		final JdbcServices jdbcServices = factory.getJdbcServices();
 		final JdbcEnvironment jdbcEnvironment = jdbcServices.getJdbcEnvironment();
 		final SqlAstTranslator<JdbcSelect> sqlAstSelectTranslator = jdbcEnvironment

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

@@ -71,7 +71,7 @@ public class SqmMutationStrategyHelper {
 			return;
 		}
 
-		entityDescriptor.visitAttributeMappings(
+		entityDescriptor.visitSubTypeAttributeMappings(
 				attributeMapping -> {
 					if ( attributeMapping instanceof PluralAttributeMapping ) {
 						cleanUpCollectionTable(

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/cte/AbstractCteMutationHandler.java

@@ -20,6 +20,7 @@ import org.hibernate.engine.jdbc.spi.JdbcServices;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.metamodel.mapping.MappingModelExpressable;
+import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.SqlExpressable;
 import org.hibernate.query.spi.DomainQueryExecutionContext;
 import org.hibernate.query.sqm.internal.DomainParameterXref;
@@ -203,6 +204,7 @@ public abstract class AbstractCteMutationHandler extends AbstractMutationHandler
 		lockOptions.setAliasSpecificLockMode( explicitDmlTargetAlias, LockMode.WRITE );
 		final JdbcSelect select = translator.translate( jdbcParameterBindings, executionContext.getQueryOptions() );
 		lockOptions.setAliasSpecificLockMode( explicitDmlTargetAlias, lockMode );
+		executionContext.getSession().autoFlushIfRequired( select.getAffectedTableNames() );
 		List<Object> list = jdbcServices.getJdbcSelectExecutor().list(
 				select,
 				jdbcParameterBindings,
@@ -230,6 +232,14 @@ public abstract class AbstractCteMutationHandler extends AbstractMutationHandler
 			List<? extends Expression> lhsExpressions,
 			CteStatement idSelectCte,
 			SessionFactoryImplementor factory) {
+		return createIdSubQueryPredicate( lhsExpressions, idSelectCte, null, factory );
+	}
+
+	protected Predicate createIdSubQueryPredicate(
+			List<? extends Expression> lhsExpressions,
+			CteStatement idSelectCte,
+			ModelPart fkModelPart,
+			SessionFactoryImplementor factory) {
 		final TableReference idSelectTableReference = new TableReference(
 				idSelectCte.getCteTable().getTableExpression(),
 				CTE_TABLE_IDENTIFIER,
@@ -238,23 +248,43 @@ public abstract class AbstractCteMutationHandler extends AbstractMutationHandler
 		);
 		final Junction predicate = new Junction( Junction.Nature.CONJUNCTION );
 		final List<CteColumn> cteColumns = idSelectCte.getCteTable().getCteColumns();
-		final int size = cteColumns.size();
+		final int size = lhsExpressions.size();
 		final QuerySpec subQuery = new QuerySpec( false, 1 );
 		subQuery.getFromClause().addRoot( new CteTableGroup( idSelectTableReference ) );
 		final SelectClause subQuerySelectClause = subQuery.getSelectClause();
-		for ( int i = 0; i < size; i++ ) {
+		if ( fkModelPart == null ) {
-			final CteColumn cteColumn = cteColumns.get( i );
+			for ( int i = 0; i < size; i++ ) {
-			subQuerySelectClause.addSqlSelection(
+				final CteColumn cteColumn = cteColumns.get( i );
-					new SqlSelectionImpl(
+				subQuerySelectClause.addSqlSelection(
-							i + 1,
+						new SqlSelectionImpl(
-							i,
+								i + 1,
-							new ColumnReference(
+								i,
-									idSelectTableReference,
+								new ColumnReference(
-									cteColumn.getColumnExpression(),
+										idSelectTableReference,
-									cteColumn.getJdbcMapping(),
+										cteColumn.getColumnExpression(),
-									factory
+										cteColumn.getJdbcMapping(),
-							)
+										factory
-					)
+								)
+						)
+				);
+			}
+		}
+		else {
+			fkModelPart.forEachSelectable(
+					(selectionIndex, selectableMapping) -> {
+						subQuerySelectClause.addSqlSelection(
+								new SqlSelectionImpl(
+										selectionIndex + 1,
+										selectionIndex,
+										new ColumnReference(
+												idSelectTableReference,
+												selectableMapping.getSelectionExpression(),
+												selectableMapping.getJdbcMapping(),
+												factory
+										)
+								)
+						);
+					}
 			);
 		}
 		final Expression lhs;

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/cte/CteDeleteHandler.java

@@ -11,10 +11,12 @@ import java.util.List;
 import java.util.Map;
 
 import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.query.sqm.internal.DomainParameterXref;
 import org.hibernate.query.sqm.mutation.internal.DeleteHandler;
 import org.hibernate.query.sqm.mutation.internal.MultiTableSqmMutationConverter;
+import org.hibernate.query.sqm.sql.internal.SqlAstQueryPartProcessingStateImpl;
 import org.hibernate.query.sqm.tree.cte.SqmCteTable;
 import org.hibernate.query.sqm.tree.delete.SqmDeleteStatement;
 import org.hibernate.query.sqm.tree.expression.SqmParameter;
@@ -27,6 +29,8 @@ import org.hibernate.sql.ast.tree.expression.ColumnReference;
 import org.hibernate.sql.ast.tree.expression.JdbcParameter;
 import org.hibernate.sql.ast.tree.from.TableGroup;
 import org.hibernate.sql.ast.tree.from.TableReference;
+import org.hibernate.sql.ast.tree.select.SelectStatement;
+import org.hibernate.sql.results.graph.basic.BasicResult;
 
 /**
  * Bulk-id delete handler that uses CTE and VALUES lists.
@@ -53,12 +57,42 @@ public class CteDeleteHandler extends AbstractCteMutationHandler implements Dele
 			Map<SqmParameter, List<JdbcParameter>> parameterResolutions,
 			SessionFactoryImplementor factory) {
 		final TableGroup updatingTableGroup = sqmConverter.getMutatingTableGroup();
-		getEntityDescriptor().visitAttributeMappings(
+		final SelectStatement idSelectStatement = (SelectStatement) idSelectCte.getCteDefinition();
+		sqmConverter.getProcessingStateStack().push(
+				new SqlAstQueryPartProcessingStateImpl(
+						idSelectStatement.getQuerySpec(),
+						sqmConverter.getCurrentProcessingState(),
+						sqmConverter.getSqlAstCreationState(),
+						sqmConverter.getCurrentClauseStack()::getCurrent
+				)
+		);
+		getEntityDescriptor().visitSubTypeAttributeMappings(
 				attribute -> {
 					if ( attribute instanceof PluralAttributeMapping ) {
 						final PluralAttributeMapping pluralAttribute = (PluralAttributeMapping) attribute;
 
 						if ( pluralAttribute.getSeparateCollectionTable() != null ) {
+							// Ensure that the FK target columns are available
+							final boolean useFkTarget = !( pluralAttribute.getKeyDescriptor()
+									.getTargetPart() instanceof EntityIdentifierMapping );
+							if ( useFkTarget ) {
+								final TableGroup mutatingTableGroup = sqmConverter.getMutatingTableGroup();
+								pluralAttribute.getKeyDescriptor().getTargetPart().applySqlSelections(
+										mutatingTableGroup.getNavigablePath(),
+										mutatingTableGroup,
+										sqmConverter,
+										(selection, jdbcMapping) -> {
+											idSelectStatement.getDomainResultDescriptors().add(
+													new BasicResult<>(
+															selection.getValuesArrayPosition(),
+															null,
+															jdbcMapping.getJavaTypeDescriptor()
+													)
+											);
+										}
+								);
+							}
+
 							// this collection has a separate collection table, meaning it is one of:
 							//		1) element-collection
 							//		2) many-to-many
@@ -68,7 +102,7 @@ public class CteDeleteHandler extends AbstractCteMutationHandler implements Dele
 							// collection table
 							final String tableExpression = pluralAttribute.getSeparateCollectionTable();
 							final CteTable dmlResultCte = new CteTable(
-									getCteTableName( tableExpression ),
+									getCteTableName( pluralAttribute ),
 									idSelectCte.getCteTable().getCteColumns(),
 									factory
 							);
@@ -85,7 +119,12 @@ public class CteDeleteHandler extends AbstractCteMutationHandler implements Dele
 							);
 							final MutationStatement dmlStatement = new DeleteStatement(
 									dmlTableReference,
-									createIdSubQueryPredicate( columnReferences, idSelectCte, factory ),
+									createIdSubQueryPredicate(
+											columnReferences,
+											idSelectCte,
+											useFkTarget ? pluralAttribute.getKeyDescriptor().getTargetPart() : null,
+											factory
+									),
 									columnReferences
 							);
 							statement.addCteStatement( new CteStatement( dmlResultCte, dmlStatement ) );
@@ -93,6 +132,7 @@ public class CteDeleteHandler extends AbstractCteMutationHandler implements Dele
 					}
 				}
 		);
+		sqmConverter.getProcessingStateStack().pop();
 
 		getEntityDescriptor().visitConstraintOrderedTables(
 				(tableExpression, tableColumnsVisitationSupplier) -> {
@@ -124,4 +164,12 @@ public class CteDeleteHandler extends AbstractCteMutationHandler implements Dele
 				}
 		);
 	}
+
+	protected String getCteTableName(PluralAttributeMapping pluralAttribute) {
+		final String hibernateEntityName = pluralAttribute.findContainingEntityMapping().getEntityName();
+		final String jpaEntityName = getSessionFactory().getJpaMetamodel().entity( hibernateEntityName ).getName();
+		return DML_RESULT_TABLE_NAME_PREFIX + jpaEntityName + "_" + pluralAttribute.getRootPathName().substring(
+				hibernateEntityName.length() + 1
+		);
+	}
 }

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/idtable/ExecuteWithIdTableHelper.java

@@ -19,6 +19,7 @@ import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.engine.transaction.spi.IsolationDelegate;
 import org.hibernate.metamodel.mapping.BasicValuedMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.query.ComparisonOperator;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.query.sqm.mutation.internal.MultiTableSqmMutationConverter;
@@ -162,6 +163,15 @@ public final class ExecuteWithIdTableHelper {
 			Function<SharedSessionContractImplementor,String> sessionUidAccess,
 			EntityMappingType entityDescriptor,
 			ExecutionContext executionContext) {
+		return createIdTableSelectQuerySpec( idTable, null, sessionUidAccess, entityDescriptor, executionContext );
+	}
+
+	public static QuerySpec createIdTableSelectQuerySpec(
+			IdTable idTable,
+			ModelPart fkModelPart,
+			Function<SharedSessionContractImplementor,String> sessionUidAccess,
+			EntityMappingType entityDescriptor,
+			ExecutionContext executionContext) {
 		final QuerySpec querySpec = new QuerySpec( false );
 
 		final TableReference idTableReference = new TableReference(
@@ -182,7 +192,7 @@ public final class ExecuteWithIdTableHelper {
 
 		querySpec.getFromClause().addRoot( idTableGroup );
 
-		applyIdTableSelections( querySpec, idTableReference, idTable, executionContext );
+		applyIdTableSelections( querySpec, idTableReference, idTable, fkModelPart, executionContext );
 		applyIdTableRestrictions( querySpec, idTableReference, idTable, sessionUidAccess, executionContext );
 
 		return querySpec;
@@ -192,34 +202,59 @@ public final class ExecuteWithIdTableHelper {
 			QuerySpec querySpec,
 			TableReference tableReference,
 			IdTable idTable,
+			ModelPart fkModelPart,
 			ExecutionContext executionContext) {
-		for ( int i = 0; i < idTable.getIdTableColumns().size(); i++ ) {
+		if ( fkModelPart == null ) {
-			final IdTableColumn idTableColumn = idTable.getIdTableColumns().get( i );
+			final int size = idTable.getEntityDescriptor().getIdentifierMapping().getJdbcTypeCount();
-			if ( idTableColumn != idTable.getSessionUidColumn() ) {
+			for ( int i = 0; i < size; i++ ) {
-				querySpec.getSelectClause().addSqlSelection(
+				final IdTableColumn idTableColumn = idTable.getIdTableColumns().get( i );
-						new SqlSelectionImpl(
+				if ( idTableColumn != idTable.getSessionUidColumn() ) {
-								i + 1,
+					querySpec.getSelectClause().addSqlSelection(
-								i,
+							new SqlSelectionImpl(
-								new ColumnReference(
+									i + 1,
-										tableReference,
+									i,
-										idTableColumn.getColumnName(),
+									new ColumnReference(
-										false,
+											tableReference,
-										null,
+											idTableColumn.getColumnName(),
-										null,
+											false,
-										idTableColumn.getJdbcMapping(),
+											null,
-										executionContext.getSession().getFactory()
+											null,
-								)
+											idTableColumn.getJdbcMapping(),
-						)
+											executionContext.getSession().getFactory()
-				);
+									)
+							)
+					);
+				}
 			}
 		}
+		else {
+			fkModelPart.forEachSelectable(
+					(i, selectableMapping) -> {
+						querySpec.getSelectClause().addSqlSelection(
+								new SqlSelectionImpl(
+										i + 1,
+										i,
+										new ColumnReference(
+												tableReference,
+												selectableMapping.getSelectionExpression(),
+												false,
+												null,
+												null,
+												selectableMapping.getJdbcMapping(),
+												executionContext.getSession().getFactory()
+										)
+								)
+						);
+					}
+			);
+		}
 	}
 
 	private static void applyIdTableRestrictions(
 			QuerySpec querySpec,
 			TableReference idTableReference,
 			IdTable idTable,
-			Function<SharedSessionContractImplementor,String> sessionUidAccess,
+			Function<SharedSessionContractImplementor, String> sessionUidAccess,
 			ExecutionContext executionContext) {
 		if ( idTable.getSessionUidColumn() != null ) {
 			querySpec.applyPredicate(

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/idtable/IdTable.java

@@ -13,8 +13,12 @@ import java.util.function.Function;
 import org.hibernate.boot.model.relational.Exportable;
 import org.hibernate.dialect.Dialect;
 import org.hibernate.mapping.Contributable;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.ModelPart;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.persister.entity.Joinable;
+import org.hibernate.sql.ast.tree.from.TableGroup;
 
 /**
  * @author Steve Ebersole
@@ -51,6 +55,32 @@ public class IdTable implements Exportable, Contributable {
 						)
 				)
 		);
+		entityDescriptor.visitSubTypeAttributeMappings(
+				attribute -> {
+					if ( attribute instanceof PluralAttributeMapping ) {
+						final PluralAttributeMapping pluralAttribute = (PluralAttributeMapping) attribute;
+
+						if ( pluralAttribute.getSeparateCollectionTable() != null ) {
+							// Ensure that the FK target columns are available
+							final ModelPart fkTarget = pluralAttribute.getKeyDescriptor().getTargetPart();
+							if ( !( fkTarget instanceof EntityIdentifierMapping ) ) {
+								fkTarget.forEachSelectable(
+										(columnIndex, selection) -> columns.add(
+												new IdTableColumn(
+														this,
+														selection.getSelectionExpression(),
+														selection.getJdbcMapping(),
+														dialect.getTypeName(
+																selection.getJdbcMapping().getJdbcTypeDescriptor()
+														)
+												)
+										)
+								);
+							}
+						}
+					}
+				}
+		);
 
 		this.dialect = dialect;
 	}

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/idtable/RestrictedDeleteExecutionDelegate.java

@@ -24,6 +24,7 @@ import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.FilterHelper;
 import org.hibernate.internal.util.MutableInteger;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.MappingModelExpressable;
@@ -448,7 +449,7 @@ public class RestrictedDeleteExecutionDelegate implements TableBasedDeleteHandle
 				executionContext
 		);
 
-		final QuerySpec idTableSubQuery = ExecuteWithIdTableHelper.createIdTableSelectQuerySpec(
+		final QuerySpec idTableIdentifierSubQuery = ExecuteWithIdTableHelper.createIdTableSelectQuerySpec(
 				idTable,
 				sessionUidAccess,
 				entityDescriptor,
@@ -459,14 +460,26 @@ public class RestrictedDeleteExecutionDelegate implements TableBasedDeleteHandle
 				entityDescriptor,
 				(tableReference, attributeMapping) -> {
 					final ForeignKeyDescriptor fkDescriptor = attributeMapping.getKeyDescriptor();
-
+					final QuerySpec idTableFkSubQuery;
+					if ( fkDescriptor.getTargetPart() instanceof EntityIdentifierMapping ) {
+						idTableFkSubQuery = idTableIdentifierSubQuery;
+					}
+					else {
+						idTableFkSubQuery = ExecuteWithIdTableHelper.createIdTableSelectQuerySpec(
+								idTable,
+								fkDescriptor.getTargetPart(),
+								sessionUidAccess,
+								entityDescriptor,
+								executionContext
+						);
+					}
 					return new InSubQueryPredicate(
 							MappingModelHelper.buildColumnReferenceExpression(
 									fkDescriptor,
 									null,
 									sessionFactory
 							),
-							idTableSubQuery,
+							idTableFkSubQuery,
 							false
 					);
 
@@ -479,7 +492,7 @@ public class RestrictedDeleteExecutionDelegate implements TableBasedDeleteHandle
 				(tableExpression, tableKeyColumnVisitationSupplier) -> deleteFromTableUsingIdTable(
 						tableExpression,
 						tableKeyColumnVisitationSupplier,
-						idTableSubQuery,
+						idTableIdentifierSubQuery,
 						executionContext
 				)
 		);

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/inline/InPredicateRestrictionProducer.java

@@ -13,6 +13,7 @@ import java.util.function.Supplier;
 
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.metamodel.mapping.BasicValuedModelPart;
+import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.SelectableConsumer;
 import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
@@ -48,6 +49,8 @@ public class InPredicateRestrictionProducer implements MatchingIdRestrictionProd
 	public InListPredicate produceRestriction(
 			List<?> matchingIdValues,
 			EntityMappingType entityDescriptor,
+			int valueIndex,
+			ModelPart valueModelPart,
 			TableReference mutatingTableReference,
 			Supplier<Consumer<SelectableConsumer>> columnsToMatchVisitationSupplier,
 			ExecutionContext executionContext) {

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/inline/InlineDeleteHandler.java

@@ -12,6 +12,10 @@ import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.internal.util.MutableInteger;
+import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
+import org.hibernate.metamodel.mapping.ModelPart;
+import org.hibernate.metamodel.mapping.SelectableConsumer;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.metamodel.mapping.SelectableConsumer;
@@ -69,13 +73,13 @@ public class InlineDeleteHandler implements DeleteHandler {
 
 	@Override
 	public int execute(DomainQueryExecutionContext executionContext) {
-		final List<Object> ids = MatchingIdSelectionHelper.selectMatchingIds(
+		final List<Object> idsAndFks = MatchingIdSelectionHelper.selectMatchingIds(
 				sqmDeleteStatement,
 				domainParameterXref,
 				executionContext
 		);
 
-		if ( ids == null || ids.isEmpty() ) {
+		if ( idsAndFks == null || idsAndFks.isEmpty() ) {
 			return 0;
 		}
 
@@ -87,8 +91,8 @@ public class InlineDeleteHandler implements DeleteHandler {
 		final JdbcParameterBindings jdbcParameterBindings = new JdbcParameterBindingsImpl( domainParameterXref.getQueryParameterCount() );
 
 		// delete from the tables
-
+		final MutableInteger valueIndexCounter = new MutableInteger();
-		entityDescriptor.visitAttributeMappings(
+		entityDescriptor.visitSubTypeAttributeMappings(
 				attribute -> {
 					if ( attribute instanceof PluralAttributeMapping ) {
 						final PluralAttributeMapping pluralAttribute = (PluralAttributeMapping) attribute;
@@ -101,18 +105,29 @@ public class InlineDeleteHandler implements DeleteHandler {
 							//
 							// in all of these cases, we should clean up the matching rows in the
 							// collection table
+							final ModelPart fkTargetPart = pluralAttribute.getKeyDescriptor().getTargetPart();
+							final int valueIndex;
+							if ( fkTargetPart instanceof EntityIdentifierMapping ) {
+								valueIndex = 0;
+							}
+							else {
+								if ( valueIndexCounter.get() == 0 ) {
+									valueIndexCounter.set( entityDescriptor.getIdentifierMapping().getJdbcTypeCount() );
+								}
+								valueIndex = valueIndexCounter.get();
+								valueIndexCounter.plus( fkTargetPart.getJdbcTypeCount() );
+							}
 
-// todo (6.0) : implement this
+							executeDelete(
-//							executeDelete(
+									pluralAttribute.getSeparateCollectionTable(),
-//									pluralAttribute.getSeparateCollectionTable(),
+									entityDescriptor,
-//									matchingIdsPredicateProducer.produceRestriction(
+									() -> fkTargetPart::forEachSelectable,
-//											ids,
+									idsAndFks,
-//											() -> columnConsumer ->  ,
+									valueIndex,
-//											executionContext
+									fkTargetPart,
-//									),
+									jdbcParameterBindings,
-//									jdbcParameterBindings,
+									executionContext
-//									executionContext
+							);
-//							);
 						}
 					}
 				}
@@ -124,14 +139,16 @@ public class InlineDeleteHandler implements DeleteHandler {
 							tableExpression,
 							entityDescriptor,
 							tableKeyColumnsVisitationSupplier,
-							ids,
+							idsAndFks,
+							0,
+							null,
 							jdbcParameterBindings,
 							executionContext
 					);
 				}
 		);
 
-		return ids.size();
+		return idsAndFks.size();
 	}
 
 	private void executeDelete(
@@ -139,6 +156,8 @@ public class InlineDeleteHandler implements DeleteHandler {
 			EntityMappingType entityDescriptor,
 			Supplier<Consumer<SelectableConsumer>> tableKeyColumnsVisitationSupplier,
 			List<Object> ids,
+			int valueIndex,
+			ModelPart valueModelPart,
 			JdbcParameterBindings jdbcParameterBindings,
 			DomainQueryExecutionContext executionContext) {
 		final TableReference targetTableReference = new TableReference(
@@ -153,6 +172,8 @@ public class InlineDeleteHandler implements DeleteHandler {
 		final Predicate matchingIdsPredicate = matchingIdsPredicateProducer.produceRestriction(
 				ids,
 				entityDescriptor,
+				valueIndex,
+				valueModelPart,
 				targetTableReference,
 				tableKeyColumnsVisitationSupplier,
 				executionContextAdapter

hibernate-core/src/main/java/org/hibernate/query/sqm/mutation/internal/inline/MatchingIdRestrictionProducer.java

@@ -10,6 +10,7 @@ import java.util.List;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.SelectableConsumer;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.sql.ast.tree.from.TableReference;
@@ -33,6 +34,8 @@ public interface MatchingIdRestrictionProducer {
 	Predicate produceRestriction(
 			List<?> matchingIdValues,
 			EntityMappingType entityDescriptor,
+			int valueIndex,
+			ModelPart valueModelPart,
 			TableReference mutatingTableReference,
 			Supplier<Consumer<SelectableConsumer>> columnsToMatchVisitationSupplier,
 			ExecutionContext executionContext);

hibernate-core/src/main/java/org/hibernate/sql/ast/tree/from/TableGroup.java

@@ -9,6 +9,8 @@ package org.hibernate.sql.ast.tree.from;
 import java.util.List;
 import java.util.function.Consumer;
 
+import org.hibernate.metamodel.mapping.EntityValuedModelPart;
+import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.ModelPartContainer;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.query.sqm.sql.internal.DomainResultProducer;
@@ -68,7 +70,15 @@ public interface TableGroup extends SqlAstNode, ColumnReferenceQualifier, SqmPat
 
 	@Override
 	default void applySqlSelections(DomainResultCreationState creationState) {
-		getModelPart().applySqlSelections(
+		final ModelPartContainer modelPart = getModelPart();
+		final ModelPart modelPartToApply;
+		if ( modelPart instanceof EntityValuedModelPart ) {
+			modelPartToApply = ( (EntityValuedModelPart) modelPart ).getEntityMappingType();
+		}
+		else {
+			modelPartToApply = modelPart;
+		}
+		modelPartToApply.applySqlSelections(
 				getNavigablePath(),
 				creationState.getSqlAstCreationState().getFromClauseAccess().findTableGroup( getNavigablePath() ),
 				creationState

hibernate-core/src/test/java/org/hibernate/orm/test/cfg/persister/GoofyPersisterClassProvider.java

@@ -700,7 +700,7 @@ public class GoofyPersisterClassProvider implements PersisterClassResolver {
 		}
 
 		@Override
-		public void visitAttributeMappings(Consumer<AttributeMapping> action) {
+		public void visitAttributeMappings(Consumer<? super AttributeMapping> action) {
 
 		}
 

hibernate-core/src/test/java/org/hibernate/orm/test/jpa/ejb3configuration/PersisterClassProviderTest.java

@@ -714,7 +714,7 @@ public class PersisterClassProviderTest {
 		}
 
 		@Override
-		public void visitAttributeMappings(Consumer<AttributeMapping> action) {
+		public void visitAttributeMappings(Consumer<? super AttributeMapping> action) {
 
 		}
 

hibernate-core/src/test/java/org/hibernate/orm/test/legacy/CustomPersister.java

@@ -815,7 +815,7 @@ public class CustomPersister implements EntityPersister {
 	}
 
 	@Override
-	public void visitAttributeMappings(Consumer<AttributeMapping> action) {
+	public void visitAttributeMappings(Consumer<? super AttributeMapping> action) {
 
 	}