HHH-16358 Make OneToMany with abstract TABLE_PER_CLASS element work again

hibernate-core/src/main/java/org/hibernate/engine/jdbc/mutation/internal/AbstractSingleMutationExecutor.java

@@ -18,7 +18,7 @@ import org.hibernate.sql.model.jdbc.JdbcValueDescriptor;
 /**
  * @author Steve Ebersole
  */
-public abstract class AbstractSingleMutationExecutor extends AbstractMutationExecutor {
+public abstract class AbstractSingleMutationExecutor extends AbstractMutationExecutor implements JdbcValueBindingsImpl.JdbcValueDescriptorAccess {
 	private final PreparableMutationOperation mutationOperation;
 	private final JdbcValueBindingsImpl valueBindings;
 
@@ -29,7 +29,7 @@ public abstract class AbstractSingleMutationExecutor extends AbstractMutationExe
 		this.valueBindings = new JdbcValueBindingsImpl(
 				mutationOperation.getMutationType(),
 				mutationOperation.getMutationTarget(),
-				this::findJdbcValueDescriptor,
+				this,
 				session
 		);
 	}
@@ -47,8 +47,14 @@ public abstract class AbstractSingleMutationExecutor extends AbstractMutationExe
 		return statementDetails;
 	}
 
-	private JdbcValueDescriptor findJdbcValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
+	@Override
-		assert mutationOperation.getTableDetails().getTableName().equals( tableName )
+	public String resolvePhysicalTableName(String tableName) {
+		return mutationOperation.getTableDetails().getTableName();
+	}
+
+	@Override
+	public JdbcValueDescriptor resolveValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
+		assert mutationOperation.getTableDetails().containsTableName( tableName )
 				: String.format( Locale.ROOT, "table names did not match : `%s` & `%s`", tableName, mutationOperation.getTableDetails().getTableName()  );
 		return mutationOperation.findValueDescriptor( columnName, usage );
 	}

hibernate-core/src/main/java/org/hibernate/engine/jdbc/mutation/internal/JdbcValueBindingsImpl.java

@@ -60,7 +60,7 @@ public class JdbcValueBindingsImpl implements JdbcValueBindings {
 			throw new UnknownParameterException( mutationType, mutationTarget, tableName, columnName, usage );
 		}
 
-		resolveBindingGroup( tableName ).bindValue( columnName, value, jdbcValueDescriptor );
+		resolveBindingGroup( jdbcValueDescriptorAccess.resolvePhysicalTableName( tableName ) ).bindValue( columnName, value, jdbcValueDescriptor );
 	}
 
 	private BindingGroup resolveBindingGroup(String tableName) {
@@ -119,8 +119,12 @@ public class JdbcValueBindingsImpl implements JdbcValueBindings {
 	/**
 	 * Access to {@link JdbcValueDescriptor} values
 	 */
-	@FunctionalInterface
 	public interface JdbcValueDescriptorAccess {
+
+		default String resolvePhysicalTableName(String tableName) {
+			return tableName;
+		}
+
 		JdbcValueDescriptor resolveValueDescriptor(String tableName, String columnName, ParameterUsage usage);
 	}
 }

hibernate-core/src/main/java/org/hibernate/engine/jdbc/mutation/internal/MutationExecutorPostInsert.java

@@ -48,7 +48,7 @@ import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER
  *
  * @author Steve Ebersole
  */
-public class MutationExecutorPostInsert implements MutationExecutor {
+public class MutationExecutorPostInsert implements MutationExecutor, JdbcValueBindingsImpl.JdbcValueDescriptorAccess {
 	protected final EntityMutationTarget mutationTarget;
 	protected final MutationOperationGroup mutationOperationGroup;
 
@@ -66,7 +66,7 @@ public class MutationExecutorPostInsert implements MutationExecutor {
 		this.valueBindings = new JdbcValueBindingsImpl(
 				MutationType.INSERT,
 				mutationTarget,
-				this::findJdbcValueDescriptor,
+				this,
 				session
 		);
 		this.mutationOperationGroup = mutationOperationGroup;
@@ -112,7 +112,8 @@ public class MutationExecutorPostInsert implements MutationExecutor {
 		return valueBindings;
 	}
 
-	private JdbcValueDescriptor findJdbcValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
+	@Override
+	public JdbcValueDescriptor resolveValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
 		final MutationOperation operation = mutationOperationGroup.getOperation( tableName );
 		if ( operation == null ) {
 			return null;

hibernate-core/src/main/java/org/hibernate/engine/jdbc/mutation/internal/MutationExecutorPostInsertSingleTable.java

@@ -11,6 +11,7 @@ import java.util.Locale;
 import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
 import org.hibernate.engine.jdbc.mutation.MutationExecutor;
 import org.hibernate.engine.jdbc.mutation.OperationResultChecker;
+import org.hibernate.engine.jdbc.mutation.ParameterUsage;
 import org.hibernate.engine.jdbc.mutation.TableInclusionChecker;
 import org.hibernate.engine.jdbc.mutation.group.PreparedStatementDetails;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -20,6 +21,7 @@ import org.hibernate.sql.model.MutationOperationGroup;
 import org.hibernate.sql.model.MutationType;
 import org.hibernate.sql.model.PreparableMutationOperation;
 import org.hibernate.sql.model.ValuesAnalysis;
+import org.hibernate.sql.model.jdbc.JdbcValueDescriptor;
 
 import static org.hibernate.engine.jdbc.mutation.internal.ModelMutationHelper.identityPreparation;
 import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER;
@@ -41,10 +43,10 @@ import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER
  *
  * @author Steve Ebersole
  */
-public class MutationExecutorPostInsertSingleTable implements MutationExecutor {
+public class MutationExecutorPostInsertSingleTable implements MutationExecutor, JdbcValueBindingsImpl.JdbcValueDescriptorAccess {
 	private final EntityMutationTarget mutationTarget;
 	private final SharedSessionContractImplementor session;
-
+	private final PreparableMutationOperation operation;
 	private final PreparedStatementDetails identityInsertStatementDetails;
 
 	private final JdbcValueBindingsImpl valueBindings;
@@ -57,20 +59,23 @@ public class MutationExecutorPostInsertSingleTable implements MutationExecutor {
 
 		assert mutationOperationGroup.getNumberOfOperations() == 1;
 
-		final PreparableMutationOperation operation = mutationOperationGroup.getOperation( mutationTarget.getIdentifierTableName() );
+		this.operation = mutationOperationGroup.getOperation( mutationTarget.getIdentifierTableName() );
 		this.identityInsertStatementDetails = identityPreparation( operation, session );
 
 		this.valueBindings = new JdbcValueBindingsImpl(
 				MutationType.INSERT,
 				mutationTarget,
-				(tableName, columnName, usage) -> {
+				this,
-					assert identityInsertStatementDetails.getMutatingTableDetails().getTableName().equals( tableName );
-					return operation.findValueDescriptor( columnName, usage );
-				},
 				session
 		);
 	}
 
+	@Override
+	public JdbcValueDescriptor resolveValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
+		assert identityInsertStatementDetails.getMutatingTableDetails().getTableName().equals( tableName );
+		return operation.findValueDescriptor( columnName, usage );
+	}
+
 	@Override
 	public JdbcValueBindings getJdbcValueBindings() {
 		return valueBindings;

hibernate-core/src/main/java/org/hibernate/engine/jdbc/mutation/internal/MutationExecutorSingleSelfExecuting.java

@@ -18,7 +18,7 @@ import org.hibernate.sql.model.jdbc.JdbcValueDescriptor;
 /**
  * @author Steve Ebersole
  */
-public class MutationExecutorSingleSelfExecuting extends AbstractMutationExecutor {
+public class MutationExecutorSingleSelfExecuting extends AbstractMutationExecutor implements JdbcValueBindingsImpl.JdbcValueDescriptorAccess {
 	private final SelfExecutingUpdateOperation operation;
 	private final JdbcValueBindingsImpl valueBindings;
 
@@ -30,14 +30,15 @@ public class MutationExecutorSingleSelfExecuting extends AbstractMutationExecuto
 		this.valueBindings = new JdbcValueBindingsImpl(
 				operation.getMutationType(),
 				operation.getMutationTarget(),
-				this::findJdbcValueDescriptor,
+				this,
 				session
 		);
 
 		prepareForNonBatchedWork( null, session );
 	}
 
-	private JdbcValueDescriptor findJdbcValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
+	@Override
+	public JdbcValueDescriptor resolveValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
 		return operation.findValueDescriptor( columnName, usage );
 	}
 

hibernate-core/src/main/java/org/hibernate/engine/jdbc/mutation/internal/MutationExecutorStandard.java

@@ -39,7 +39,7 @@ import static org.hibernate.internal.util.collections.CollectionHelper.isNotEmpt
  *
  * @author Steve Ebersole
  */
-public class MutationExecutorStandard extends AbstractMutationExecutor {
+public class MutationExecutorStandard extends AbstractMutationExecutor implements JdbcValueBindingsImpl.JdbcValueDescriptorAccess {
 	private final MutationOperationGroup mutationOperationGroup;
 
 	/**
@@ -156,7 +156,7 @@ public class MutationExecutorStandard extends AbstractMutationExecutor {
 		this.valueBindings = new JdbcValueBindingsImpl(
 				mutationOperationGroup.getMutationType(),
 				mutationOperationGroup.getMutationTarget(),
-				this::findJdbcValueDescriptor,
+				this,
 				session
 		);
 
@@ -174,7 +174,8 @@ public class MutationExecutorStandard extends AbstractMutationExecutor {
 		return valueBindings;
 	}
 
-	private JdbcValueDescriptor findJdbcValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
+	@Override
+	public JdbcValueDescriptor resolveValueDescriptor(String tableName, String columnName, ParameterUsage usage) {
 		return mutationOperationGroup.getOperation( tableName ).findValueDescriptor( columnName, usage );
 	}
 

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

@@ -73,6 +73,7 @@ import org.hibernate.metamodel.mapping.SelectablePath;
 import org.hibernate.metamodel.mapping.VirtualModelPart;
 import org.hibernate.metamodel.model.domain.NavigableRole;
 import org.hibernate.metamodel.spi.RuntimeModelCreationContext;
+import org.hibernate.persister.collection.AbstractCollectionPersister;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.persister.collection.QueryableCollection;
 import org.hibernate.persister.collection.SQLLoadableCollection;
@@ -737,6 +738,7 @@ public class MappingModelCreationHelper {
 		final String lhsPropertyName = collectionDescriptor.getCollectionType().getLHSPropertyName();
 		final boolean isReferenceToPrimaryKey = lhsPropertyName == null;
 		final ManagedMappingType keyDeclaringType;
+		final String collectionTableName = ((AbstractCollectionPersister) collectionDescriptor).getTableName();
 
 		if ( collectionDescriptor.getElementType().isEntityType() ) {
 			keyDeclaringType = ( (QueryableCollection) collectionDescriptor ).getElementPersister();
@@ -762,7 +764,7 @@ public class MappingModelCreationHelper {
 
 			final BasicValuedModelPart simpleFkTargetPart = (BasicValuedModelPart) fkTargetPart;
 
-			final String keyTableExpression = getTableIdentifierExpression( bootValueMappingKey.getTable(), creationProcess );
+			final String keyTableExpression = collectionTableName;//getTableIdentifierExpression( bootValueMappingKey.getTable(), creationProcess );
 			final SelectableMapping keySelectableMapping = SelectableMappingImpl.from(
 					keyTableExpression,
 					bootValueMappingKey.getSelectables().get(0),
@@ -791,6 +793,7 @@ public class MappingModelCreationHelper {
 					bootValueMapping,
 					keyDeclaringType,
 					collectionDescriptor.getAttributeMapping(),
+					collectionTableName,
 					false,
 					bootValueMappingKey.getColumnInsertability(),
 					bootValueMappingKey.getColumnUpdateability(),
@@ -1066,13 +1069,37 @@ public class MappingModelCreationHelper {
 			boolean[] updateable,
 			Dialect dialect,
 			MappingModelCreationProcess creationProcess) {
+		return buildEmbeddableForeignKeyDescriptor(
+				embeddableValuedModelPart,
+				bootValueMapping,
+				keyDeclaringType,
+				keyDeclaringTableGroupProducer,
+				null,
+				inverse,
+				insertable,
+				updateable,
+				dialect,
+				creationProcess
+		);
+	}
+
+	private static EmbeddedForeignKeyDescriptor buildEmbeddableForeignKeyDescriptor(
+			EmbeddableValuedModelPart embeddableValuedModelPart,
+			Value bootValueMapping,
+			ManagedMappingType keyDeclaringType,
+			TableGroupProducer keyDeclaringTableGroupProducer,
+			String keyTableExpression,
+			boolean inverse,
+			boolean[] insertable,
+			boolean[] updateable,
+			Dialect dialect,
+			MappingModelCreationProcess creationProcess) {
 		final boolean hasConstraint;
 		final SelectableMappings keySelectableMappings;
-		final String keyTableExpression;
 		if ( bootValueMapping instanceof Collection ) {
 			final Collection collectionBootValueMapping = (Collection) bootValueMapping;
 			hasConstraint = ((SimpleValue) collectionBootValueMapping.getKey()).isConstrained();
-			keyTableExpression = getTableIdentifierExpression(
+			keyTableExpression = keyTableExpression != null ? keyTableExpression : getTableIdentifierExpression(
 					collectionBootValueMapping.getCollectionTable(),
 					creationProcess
 			);
@@ -1096,7 +1123,7 @@ public class MappingModelCreationHelper {
 			else {
 				hasConstraint = ((SimpleValue) bootValueMapping).isConstrained();
 			}
-			keyTableExpression = getTableIdentifierExpression(
+			keyTableExpression = keyTableExpression != null ? keyTableExpression : getTableIdentifierExpression(
 					bootValueMapping.getTable(),
 					creationProcess
 			);

hibernate-core/src/main/java/org/hibernate/persister/collection/AbstractCollectionPersister.java

@@ -242,7 +242,7 @@ public abstract class AbstractCollectionPersister
 	private final String manyToManyWhereString;
 	private final String manyToManyWhereTemplate;
 
-	private final Serializable[] spaces;
+	private final String[] spaces;
 
 	private final Map<String,String[]> collectionPropertyColumnAliases = new HashMap<>();
 
@@ -304,7 +304,6 @@ public abstract class AbstractCollectionPersister
 
 		int spacesSize = 1 + collectionBootDescriptor.getSynchronizedTables().size();
 		spaces = new String[spacesSize];
-		spaces[0] = qualifiedTableName;
 		Iterator<String> tables = collectionBootDescriptor.getSynchronizedTables().iterator();
 		for ( int i = 1; i < spacesSize; i++ ) {
 			spaces[i] = tables.next();
@@ -363,6 +362,8 @@ public abstract class AbstractCollectionPersister
 		else {
 			elementPersister = null;
 		}
+		// Defer this after the element persister was determined, because it is needed in OneToManyPersister#getTableName()
+		spaces[0] = getTableName();
 
 		int elementSpan = elementBootDescriptor.getColumnSpan();
 		elementColumnAliases = new String[elementSpan];
@@ -602,7 +603,7 @@ public abstract class AbstractCollectionPersister
 			}
 		}
 
-		tableMapping = buildCollectionTableMapping( collectionBootDescriptor, qualifiedTableName );
+		tableMapping = buildCollectionTableMapping( collectionBootDescriptor, getTableName(), getCollectionSpaces() );
 	}
 
 	private BeforeExecutionGenerator createGenerator(RuntimeModelCreationContext context, IdentifierCollection collection) {
@@ -1330,7 +1331,7 @@ public abstract class AbstractCollectionPersister
 	}
 
 	@Override
-	public Serializable[] getCollectionSpaces() {
+	public String[] getCollectionSpaces() {
 		return spaces;
 	}
 
@@ -1631,9 +1632,11 @@ public abstract class AbstractCollectionPersister
 
 	private static CollectionTableMapping buildCollectionTableMapping(
 			Collection collectionBootDescriptor,
-			String qualifiedTableName) {
+			String qualifiedTableName,
+			String[] spaces) {
 		return new CollectionTableMapping(
 				qualifiedTableName,
+				spaces,
 				!collectionBootDescriptor.isOneToMany(),
 				collectionBootDescriptor.isInverse(),
 				new MutationDetails(

hibernate-core/src/main/java/org/hibernate/persister/collection/OneToManyPersister.java

@@ -41,18 +41,23 @@ import org.hibernate.persister.collection.mutation.CollectionTableMapping;
 import org.hibernate.persister.collection.mutation.DeleteRowsCoordinator;
 import org.hibernate.persister.collection.mutation.DeleteRowsCoordinatorNoOp;
 import org.hibernate.persister.collection.mutation.DeleteRowsCoordinatorStandard;
+import org.hibernate.persister.collection.mutation.DeleteRowsCoordinatorTablePerSubclass;
 import org.hibernate.persister.collection.mutation.InsertRowsCoordinator;
 import org.hibernate.persister.collection.mutation.InsertRowsCoordinatorNoOp;
 import org.hibernate.persister.collection.mutation.InsertRowsCoordinatorStandard;
+import org.hibernate.persister.collection.mutation.InsertRowsCoordinatorTablePerSubclass;
 import org.hibernate.persister.collection.mutation.OperationProducer;
 import org.hibernate.persister.collection.mutation.RemoveCoordinator;
 import org.hibernate.persister.collection.mutation.RemoveCoordinatorNoOp;
 import org.hibernate.persister.collection.mutation.RemoveCoordinatorStandard;
+import org.hibernate.persister.collection.mutation.RemoveCoordinatorTablePerSubclass;
 import org.hibernate.persister.collection.mutation.RowMutationOperations;
 import org.hibernate.persister.collection.mutation.UpdateRowsCoordinator;
 import org.hibernate.persister.collection.mutation.UpdateRowsCoordinatorNoOp;
 import org.hibernate.persister.collection.mutation.UpdateRowsCoordinatorOneToMany;
+import org.hibernate.persister.collection.mutation.UpdateRowsCoordinatorTablePerSubclass;
 import org.hibernate.persister.entity.Joinable;
+import org.hibernate.persister.entity.UnionSubclassEntityPersister;
 import org.hibernate.persister.spi.PersisterCreationContext;
 import org.hibernate.sql.ast.SqlAstTranslator;
 import org.hibernate.sql.ast.spi.SqlAstCreationState;
@@ -446,6 +451,10 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			return new InsertRowsCoordinatorNoOp( this );
 		}
 
+		if ( getElementPersisterInternal() != null && getElementPersisterInternal().hasSubclasses()
+				&& getElementPersisterInternal() instanceof UnionSubclassEntityPersister ) {
+			return new InsertRowsCoordinatorTablePerSubclass( this, rowMutationOperations );
+		}
 		return new InsertRowsCoordinatorStandard( this, rowMutationOperations );
 	}
 
@@ -460,6 +469,10 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			return new UpdateRowsCoordinatorNoOp( this );
 		}
 
+		if ( getElementPersisterInternal() != null && getElementPersisterInternal().hasSubclasses()
+				&& getElementPersisterInternal() instanceof UnionSubclassEntityPersister ) {
+			return new UpdateRowsCoordinatorTablePerSubclass( this, rowMutationOperations, getFactory() );
+		}
 		return new UpdateRowsCoordinatorOneToMany( this, getRowMutationOperations(), getFactory() );
 	}
 
@@ -474,6 +487,11 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			return new DeleteRowsCoordinatorNoOp( this );
 		}
 
+
+		if ( getElementPersisterInternal() != null && getElementPersisterInternal().hasSubclasses()
+				&& getElementPersisterInternal() instanceof UnionSubclassEntityPersister ) {
+			return new DeleteRowsCoordinatorTablePerSubclass( this, rowMutationOperations, false );
+		}
 		return new DeleteRowsCoordinatorStandard(
 				this,
 				rowMutationOperations,
@@ -493,6 +511,10 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			return new RemoveCoordinatorNoOp( this );
 		}
 
+		if ( getElementPersisterInternal() != null && getElementPersisterInternal().hasSubclasses()
+				&& getElementPersisterInternal() instanceof UnionSubclassEntityPersister ) {
+			return new RemoveCoordinatorTablePerSubclass( this, this::buildDeleteAllOperation );
+		}
 		return new RemoveCoordinatorStandard( this, this::buildDeleteAllOperation );
 	}
 
@@ -645,7 +667,7 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 
 		final EntityCollectionPart elementDescriptor = (EntityCollectionPart) attributeMapping.getElementDescriptor();
 		final EntityMappingType elementType = elementDescriptor.getAssociatedEntityMappingType();
-		assert tableReference.getTableName().equals( elementType.getIdentifierMapping().getContainingTableExpression() );
+		assert elementType.containsTableReference( tableReference.getTableName() );
 		updateBuilder.addKeyRestrictionsLeniently( elementType.getIdentifierMapping() );
 		return (TableUpdate<JdbcMutationOperation>) updateBuilder.buildMutation();
 	}

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/CollectionTableMapping.java

@@ -13,6 +13,7 @@ import org.hibernate.sql.model.TableMapping;
  */
 public class CollectionTableMapping implements TableMapping {
 	private final String tableName;
+	private final String[] spaces;
 	private final boolean isJoinTable;
 	private final boolean isInverse;
 	private final MutationDetails insertDetails;
@@ -27,6 +28,7 @@ public class CollectionTableMapping implements TableMapping {
 
 	public CollectionTableMapping(
 			String tableName,
+			String[] spaces,
 			boolean isJoinTable,
 			boolean isInverse,
 			MutationDetails insertDetails,
@@ -35,6 +37,7 @@ public class CollectionTableMapping implements TableMapping {
 			MutationDetails deleteAllDetails,
 			MutationDetails deleteRowDetails) {
 		this.tableName = tableName;
+		this.spaces = spaces;
 		this.isJoinTable = isJoinTable;
 		this.isInverse = isInverse;
 		this.insertDetails = insertDetails;
@@ -49,6 +52,24 @@ public class CollectionTableMapping implements TableMapping {
 		return tableName;
 	}
 
+	public String[] getSpaces() {
+		return spaces;
+	}
+
+	@Override
+	public boolean containsTableName(String tableName) {
+		if ( this.tableName.equals( tableName ) ) {
+			return true;
+		}
+		for ( String space : spaces ) {
+			if ( space.equals( tableName ) ) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+
 	@Override
 	public KeyDetails getKeyDetails() {
 		// todo (tuple-cleanup) : implement this

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/DeleteRowsCoordinatorTablePerSubclass.java

@@ -0,0 +1,172 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html.
+ */
+package org.hibernate.persister.collection.mutation;
+
+import java.util.Iterator;
+import java.util.function.Supplier;
+
+import org.hibernate.collection.spi.PersistentCollection;
+import org.hibernate.engine.jdbc.batch.internal.BasicBatchKey;
+import org.hibernate.engine.jdbc.batch.spi.BatchKey;
+import org.hibernate.engine.jdbc.mutation.MutationExecutor;
+import org.hibernate.engine.jdbc.mutation.spi.BatchKeyAccess;
+import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
+import org.hibernate.engine.spi.EntityEntry;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
+import org.hibernate.persister.collection.CollectionPersister;
+import org.hibernate.persister.collection.OneToManyPersister;
+import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.sql.model.MutationType;
+import org.hibernate.sql.model.internal.MutationOperationGroupSingle;
+import org.hibernate.sql.model.jdbc.JdbcMutationOperation;
+
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER;
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER_DEBUG_ENABLED;
+
+/**
+ * OneToMany delete coordinator if the element is a {@link org.hibernate.persister.entity.UnionSubclassEntityPersister}.
+ */
+public class DeleteRowsCoordinatorTablePerSubclass implements DeleteRowsCoordinator {
+	private final CollectionMutationTarget mutationTarget;
+	private final RowMutationOperations rowMutationOperations;
+	private final boolean deleteByIndex;
+
+	private final SubclassEntry[] subclassEntries;
+
+	public DeleteRowsCoordinatorTablePerSubclass(
+			OneToManyPersister mutationTarget,
+			RowMutationOperations rowMutationOperations,
+			boolean deleteByIndex) {
+		this.mutationTarget = mutationTarget;
+		this.rowMutationOperations = rowMutationOperations;
+		this.deleteByIndex = deleteByIndex;
+		this.subclassEntries = new SubclassEntry[mutationTarget.getElementPersister().getRootEntityDescriptor().getSubclassEntityNames().size()];
+	}
+
+	@Override
+	public CollectionMutationTarget getMutationTarget() {
+		return mutationTarget;
+	}
+
+	@Override
+	public void deleteRows(PersistentCollection<?> collection, Object key, SharedSessionContractImplementor session) {
+		if ( MODEL_MUTATION_LOGGER_DEBUG_ENABLED ) {
+			MODEL_MUTATION_LOGGER.debugf(
+					"Deleting removed collection rows - %s : %s",
+					mutationTarget.getRolePath(),
+					key
+			);
+		}
+
+		final MutationExecutorService mutationExecutorService = session
+				.getFactory()
+				.getServiceRegistry()
+				.getService( MutationExecutorService.class );
+
+		final PluralAttributeMapping pluralAttribute = mutationTarget.getTargetPart();
+		final CollectionPersister collectionDescriptor = pluralAttribute.getCollectionDescriptor();
+
+		final Iterator<?> deletes = collection.getDeletes( collectionDescriptor, !deleteByIndex );
+		if ( !deletes.hasNext() ) {
+			MODEL_MUTATION_LOGGER.debug( "No rows to delete" );
+			return;
+		}
+		final MutationExecutor[] executors = new MutationExecutor[subclassEntries.length];
+		try {
+			int deletionCount = 0;
+
+			final RowMutationOperations.Restrictions restrictions = rowMutationOperations.getDeleteRowRestrictions();
+
+			while ( deletes.hasNext() ) {
+				final Object removal = deletes.next();
+				final EntityEntry entityEntry = session.getPersistenceContextInternal().getEntry( removal );
+				final int subclassId = entityEntry.getPersister().getSubclassId();
+				final MutationExecutor mutationExecutor;
+				if ( executors[subclassId] == null ) {
+					final SubclassEntry subclassEntry = getSubclassEntry( entityEntry.getPersister() );
+					mutationExecutor = executors[subclassId] = mutationExecutorService.createExecutor(
+							subclassEntry.batchKeySupplier,
+							subclassEntry.operationGroup,
+							session
+					);
+				}
+				else {
+					mutationExecutor = executors[subclassId];
+				}
+
+				restrictions.applyRestrictions(
+						collection,
+						key,
+						removal,
+						deletionCount,
+						session,
+						mutationExecutor.getJdbcValueBindings()
+				);
+
+				mutationExecutor.execute( removal, null, null, null, session );
+
+				deletionCount++;
+			}
+
+			MODEL_MUTATION_LOGGER.debugf( "Done deleting `%s` collection rows : %s", deletionCount, mutationTarget.getRolePath() );
+		}
+		finally {
+			for ( MutationExecutor executor : executors ) {
+				if ( executor != null ) {
+					executor.release();
+				}
+			}
+		}
+	}
+
+	private SubclassEntry getSubclassEntry(EntityPersister elementPersister) {
+		final int subclassId = elementPersister.getSubclassId();
+		final SubclassEntry subclassEntry = subclassEntries[subclassId];
+		if ( subclassEntry != null ) {
+			return subclassEntry;
+		}
+		final BasicBatchKey basicBatchKey = new BasicBatchKey( mutationTarget.getRolePath() + "#DELETE#" + subclassId );
+		return subclassEntries[subclassId] = new SubclassEntry(
+				() -> basicBatchKey,
+				createOperationGroup( elementPersister )
+		);
+	}
+
+	private MutationOperationGroupSingle createOperationGroup(EntityPersister elementPersister) {
+		assert mutationTarget.getTargetPart() != null;
+		assert mutationTarget.getTargetPart().getKeyDescriptor() != null;
+
+		final CollectionTableMapping collectionTableMapping = mutationTarget.getCollectionTableMapping();
+		final JdbcMutationOperation operation = rowMutationOperations.getDeleteRowOperation(
+				new CollectionTableMapping(
+						elementPersister.getMappedTableDetails().getTableName(),
+						collectionTableMapping.getSpaces(),
+						collectionTableMapping.isJoinTable(),
+						collectionTableMapping.isInverse(),
+						collectionTableMapping.getInsertDetails(),
+						collectionTableMapping.getUpdateDetails(),
+						collectionTableMapping.isCascadeDeleteEnabled(),
+						collectionTableMapping.getDeleteDetails(),
+						collectionTableMapping.getDeleteRowDetails()
+				)
+		);
+		return new MutationOperationGroupSingle( MutationType.DELETE, mutationTarget, operation );
+	}
+
+	private static class SubclassEntry {
+
+		private final BatchKeyAccess batchKeySupplier;
+
+		private final MutationOperationGroupSingle operationGroup;
+
+		public SubclassEntry(BatchKeyAccess batchKeySupplier, MutationOperationGroupSingle operationGroup) {
+			this.batchKeySupplier = batchKeySupplier;
+			this.operationGroup = operationGroup;
+		}
+	}
+}

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/InsertRowsCoordinatorTablePerSubclass.java

@@ -0,0 +1,181 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html.
+ */
+package org.hibernate.persister.collection.mutation;
+
+import java.util.Iterator;
+
+import org.hibernate.collection.spi.PersistentCollection;
+import org.hibernate.engine.jdbc.batch.internal.BasicBatchKey;
+import org.hibernate.engine.jdbc.mutation.MutationExecutor;
+import org.hibernate.engine.jdbc.mutation.spi.BatchKeyAccess;
+import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
+import org.hibernate.engine.spi.EntityEntry;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
+import org.hibernate.persister.collection.CollectionPersister;
+import org.hibernate.persister.collection.OneToManyPersister;
+import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.sql.model.MutationType;
+import org.hibernate.sql.model.internal.MutationOperationGroupSingle;
+import org.hibernate.sql.model.jdbc.JdbcMutationOperation;
+
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER;
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER_DEBUG_ENABLED;
+
+/**
+ * OneToMany insert coordinator if the element is a {@link org.hibernate.persister.entity.UnionSubclassEntityPersister}.
+ */
+public class InsertRowsCoordinatorTablePerSubclass implements InsertRowsCoordinator {
+	private final CollectionMutationTarget mutationTarget;
+	private final RowMutationOperations rowMutationOperations;
+
+	private final SubclassEntry[] subclassEntries;
+
+	public InsertRowsCoordinatorTablePerSubclass(
+			OneToManyPersister mutationTarget,
+			RowMutationOperations rowMutationOperations) {
+		this.mutationTarget = mutationTarget;
+		this.rowMutationOperations = rowMutationOperations;
+		this.subclassEntries = new SubclassEntry[mutationTarget.getElementPersister().getRootEntityDescriptor().getSubclassEntityNames().size()];
+	}
+
+	@Override
+	public String toString() {
+		return "InsertRowsCoordinator(" + mutationTarget.getRolePath() + ")";
+	}
+
+	@Override
+	public CollectionMutationTarget getMutationTarget() {
+		return mutationTarget;
+	}
+
+	@Override
+	public void insertRows(
+			PersistentCollection<?> collection,
+			Object id,
+			EntryFilter entryChecker,
+			SharedSessionContractImplementor session) {
+		if ( MODEL_MUTATION_LOGGER_DEBUG_ENABLED ) {
+			MODEL_MUTATION_LOGGER.debugf(
+					"Inserting collection rows - %s : %s",
+					mutationTarget.getRolePath(),
+					id
+			);
+		}
+
+		final PluralAttributeMapping pluralAttribute = mutationTarget.getTargetPart();
+		final CollectionPersister collectionDescriptor = pluralAttribute.getCollectionDescriptor();
+
+		final MutationExecutorService mutationExecutorService = session
+				.getFactory()
+				.getServiceRegistry()
+				.getService( MutationExecutorService.class );
+
+		final Iterator<?> entries = collection.entries( collectionDescriptor );
+		collection.preInsert( collectionDescriptor );
+		if ( !entries.hasNext() ) {
+			MODEL_MUTATION_LOGGER.debugf(
+					"No collection rows to insert - %s : %s",
+					mutationTarget.getRolePath(),
+					id
+			);
+			return;
+		}
+		final MutationExecutor[] executors = new MutationExecutor[subclassEntries.length];
+		try {
+			int entryCount = 0;
+			while ( entries.hasNext() ) {
+				final Object entry = entries.next();
+
+				if ( entryChecker == null || entryChecker.include( entry, entryCount, collection, pluralAttribute ) ) {
+					final EntityEntry entityEntry = session.getPersistenceContextInternal().getEntry( entry );
+					final int subclassId = entityEntry.getPersister().getSubclassId();
+					final MutationExecutor mutationExecutor;
+					if ( executors[subclassId] == null ) {
+						final SubclassEntry subclassEntry = getSubclassEntry( entityEntry.getPersister() );
+						mutationExecutor = executors[subclassId] = mutationExecutorService.createExecutor(
+								subclassEntry.batchKeySupplier,
+								subclassEntry.operationGroup,
+								session
+						);
+					}
+					else {
+						mutationExecutor = executors[subclassId];
+					}
+					// if the entry is included, perform the "insert"
+					rowMutationOperations.getInsertRowValues().applyValues(
+							collection,
+							id,
+							entry,
+							entryCount,
+							session,
+							mutationExecutor.getJdbcValueBindings()
+					);
+					mutationExecutor.execute( entry, null, null, null, session );
+				}
+
+				entryCount++;
+			}
+
+			MODEL_MUTATION_LOGGER.debugf( "Done inserting `%s` collection rows : %s", entryCount, mutationTarget.getRolePath() );
+
+		}
+		finally {
+			for ( MutationExecutor executor : executors ) {
+				if ( executor != null ) {
+					executor.release();
+				}
+			}
+		}
+	}
+
+	private SubclassEntry getSubclassEntry(EntityPersister elementPersister) {
+		final int subclassId = elementPersister.getSubclassId();
+		final SubclassEntry subclassEntry = subclassEntries[subclassId];
+		if ( subclassEntry != null ) {
+			return subclassEntry;
+		}
+		final BasicBatchKey basicBatchKey = new BasicBatchKey( mutationTarget.getRolePath() + "#INSERT#" + subclassId );
+		return subclassEntries[subclassId] = new SubclassEntry(
+				() -> basicBatchKey,
+				createOperationGroup( elementPersister )
+		);
+	}
+
+	private MutationOperationGroupSingle createOperationGroup(EntityPersister elementPersister) {
+		assert mutationTarget.getTargetPart() != null;
+		assert mutationTarget.getTargetPart().getKeyDescriptor() != null;
+
+		final CollectionTableMapping collectionTableMapping = mutationTarget.getCollectionTableMapping();
+		final JdbcMutationOperation operation = rowMutationOperations.getInsertRowOperation(
+				new CollectionTableMapping(
+						elementPersister.getMappedTableDetails().getTableName(),
+						collectionTableMapping.getSpaces(),
+						collectionTableMapping.isJoinTable(),
+						collectionTableMapping.isInverse(),
+						collectionTableMapping.getInsertDetails(),
+						collectionTableMapping.getUpdateDetails(),
+						collectionTableMapping.isCascadeDeleteEnabled(),
+						collectionTableMapping.getDeleteDetails(),
+						collectionTableMapping.getDeleteRowDetails()
+				)
+		);
+		return new MutationOperationGroupSingle( MutationType.INSERT, mutationTarget, operation );
+	}
+
+	private static class SubclassEntry {
+
+		private final BatchKeyAccess batchKeySupplier;
+
+		private final MutationOperationGroupSingle operationGroup;
+
+		public SubclassEntry(BatchKeyAccess batchKeySupplier, MutationOperationGroupSingle operationGroup) {
+			this.batchKeySupplier = batchKeySupplier;
+			this.operationGroup = operationGroup;
+		}
+	}
+}

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/RemoveCoordinatorTablePerSubclass.java

@@ -0,0 +1,157 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html.
+ */
+package org.hibernate.persister.collection.mutation;
+
+import java.util.Collection;
+
+import org.hibernate.engine.jdbc.mutation.MutationExecutor;
+import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
+import org.hibernate.persister.collection.OneToManyPersister;
+import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.sql.model.MutationType;
+import org.hibernate.sql.model.ast.MutatingTableReference;
+import org.hibernate.sql.model.internal.MutationOperationGroupSingle;
+
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER;
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER_DEBUG_ENABLED;
+import static org.hibernate.sql.model.ModelMutationLogging.MODEL_MUTATION_LOGGER_TRACE_ENABLED;
+
+/**
+ * OneToMany remove coordinator if the element is a {@link org.hibernate.persister.entity.UnionSubclassEntityPersister}.
+ */
+public class RemoveCoordinatorTablePerSubclass implements RemoveCoordinator {
+	private final OneToManyPersister mutationTarget;
+	private final OperationProducer operationProducer;
+
+	private MutationOperationGroupSingle[] operationGroups;
+
+	/**
+	 * Creates the coordinator.
+	 *
+	 * @implNote We pass a Supplier here and lazily create the operation-group because
+	 * of timing (chicken-egg) back on the persister.
+	 */
+	public RemoveCoordinatorTablePerSubclass(
+			OneToManyPersister mutationTarget,
+			OperationProducer operationProducer) {
+		this.mutationTarget = mutationTarget;
+		this.operationProducer = operationProducer;
+	}
+
+	@Override
+	public String toString() {
+		return "RemoveCoordinator(" + mutationTarget.getRolePath() + ")";
+	}
+
+	@Override
+	public CollectionMutationTarget getMutationTarget() {
+		return mutationTarget;
+	}
+
+	@Override
+	public String getSqlString() {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public void deleteAllRows(Object key, SharedSessionContractImplementor session) {
+		if ( MODEL_MUTATION_LOGGER_DEBUG_ENABLED ) {
+			MODEL_MUTATION_LOGGER.debugf(
+					"Deleting collection - %s : %s",
+					mutationTarget.getRolePath(),
+					key
+			);
+		}
+
+		MutationOperationGroupSingle[] operationGroups = this.operationGroups;
+		if ( operationGroups == null ) {
+			// delayed creation of the operation-group
+			operationGroups = this.operationGroups = buildOperationGroups();
+		}
+
+		final MutationExecutorService mutationExecutorService = session
+				.getFactory()
+				.getServiceRegistry()
+				.getService( MutationExecutorService.class );
+		final ForeignKeyDescriptor fkDescriptor = mutationTarget.getTargetPart().getKeyDescriptor();
+
+		for ( MutationOperationGroupSingle operationGroup : operationGroups ) {
+			final MutationExecutor mutationExecutor = mutationExecutorService.createExecutor(
+					() -> null,
+					operationGroup,
+					session
+			);
+
+			try {
+				fkDescriptor.getKeyPart().decompose(
+						key,
+						0,
+						mutationExecutor.getJdbcValueBindings(),
+						null,
+						RowMutationOperations.DEFAULT_RESTRICTOR,
+						session
+				);
+
+				mutationExecutor.execute(
+						key,
+						null,
+						null,
+						null,
+						session
+				);
+			}
+			finally {
+				mutationExecutor.release();
+			}
+		}
+	}
+
+	private MutationOperationGroupSingle[] buildOperationGroups() {
+		final Collection<EntityMappingType> subMappingTypes = mutationTarget.getElementPersister()
+				.getRootEntityDescriptor()
+				.getSubMappingTypes();
+		final MutationOperationGroupSingle[] operationGroups = new MutationOperationGroupSingle[subMappingTypes.size()];
+		int i = 0;
+		for ( EntityMappingType subMappingType : subMappingTypes ) {
+			operationGroups[i++] = buildOperationGroup( subMappingType.getEntityPersister() );
+		}
+		return operationGroups;
+	}
+
+	private MutationOperationGroupSingle buildOperationGroup(EntityPersister elementPersister) {
+		assert mutationTarget.getTargetPart() != null;
+		assert mutationTarget.getTargetPart().getKeyDescriptor() != null;
+
+		if ( MODEL_MUTATION_LOGGER_TRACE_ENABLED ) {
+			MODEL_MUTATION_LOGGER.tracef( "Starting RemoveCoordinator#buildOperationGroup - %s", mutationTarget.getRolePath() );
+		}
+
+		final CollectionTableMapping collectionTableMapping = mutationTarget.getCollectionTableMapping();
+		final MutatingTableReference tableReference = new MutatingTableReference(
+				new CollectionTableMapping(
+						elementPersister.getMappedTableDetails().getTableName(),
+						collectionTableMapping.getSpaces(),
+						collectionTableMapping.isJoinTable(),
+						collectionTableMapping.isInverse(),
+						collectionTableMapping.getInsertDetails(),
+						collectionTableMapping.getUpdateDetails(),
+						collectionTableMapping.isCascadeDeleteEnabled(),
+						collectionTableMapping.getDeleteDetails(),
+						collectionTableMapping.getDeleteRowDetails()
+				)
+		);
+
+		return new MutationOperationGroupSingle(
+				MutationType.DELETE,
+				mutationTarget,
+				operationProducer.createOperation( tableReference )
+		);
+	}
+}

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/RowMutationOperations.java

@@ -13,6 +13,7 @@ import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.NullnessHelper;
 import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.sql.model.MutationOperation;
+import org.hibernate.sql.model.TableMapping;
 import org.hibernate.sql.model.ast.MutatingTableReference;
 import org.hibernate.sql.model.jdbc.JdbcMutationOperation;
 
@@ -104,6 +105,15 @@ public class RowMutationOperations {
 		return local;
 	}
 
+	public JdbcMutationOperation getInsertRowOperation(TableMapping tableMapping) {
+		if ( !hasInsertRow() ) {
+			return null;
+		}
+
+		final MutatingTableReference tableReference = new MutatingTableReference( tableMapping );
+		return insertRowOperationProducer.createOperation( tableReference );
+	}
+
 
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// update row
@@ -160,6 +170,15 @@ public class RowMutationOperations {
 		return local;
 	}
 
+	public JdbcMutationOperation getDeleteRowOperation(TableMapping tableMapping) {
+		if ( !hasInsertRow() ) {
+			return null;
+		}
+
+		final MutatingTableReference tableReference = new MutatingTableReference( tableMapping );
+		return deleteRowOperationProducer.createOperation( tableReference );
+	}
+
 
 	@FunctionalInterface
 	public interface Restrictions {

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/UpdateRowsCoordinatorOneToMany.java

@@ -53,11 +53,14 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 	}
 
 	private void deleteRows(Object key, PersistentCollection<?> collection, SharedSessionContractImplementor session) {
-		final MutationOperationGroupSingle operationGroup = resolveDeleteGroup();
-
 		final PluralAttributeMapping attributeMapping = getMutationTarget().getTargetPart();
 		final CollectionPersister collectionDescriptor = attributeMapping.getCollectionDescriptor();
+		final Iterator<?> entries = collection.entries( collectionDescriptor );
+		if ( !entries.hasNext() ) {
+			return;
+		}
 
+		final MutationOperationGroupSingle operationGroup = resolveDeleteGroup();
 		final MutationExecutorService mutationExecutorService = session
 				.getFactory()
 				.getServiceRegistry()
@@ -71,7 +74,6 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 		try {
 			final JdbcValueBindings jdbcValueBindings = mutationExecutor.getJdbcValueBindings();
 
-			final Iterator<?> entries = collection.entries( collectionDescriptor );
 			int entryPosition = -1;
 
 			while ( entries.hasNext() ) {
@@ -81,17 +83,18 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 				if ( !collection.needsUpdating( entry, entryPosition, attributeMapping ) ) {
 					continue;
 				}
+				final Object entryToUpdate = collection.getSnapshotElement( entry, entryPosition );
 
 				rowMutationOperations.getDeleteRowRestrictions().applyRestrictions(
 						collection,
 						key,
-						entry,
+						entryToUpdate,
 						entryPosition,
 						session,
 						jdbcValueBindings
 				);
 
-				mutationExecutor.execute( entry, null, null, null, session );
+				mutationExecutor.execute( entryToUpdate, null, null, null, session );
 			}
 		}
 		finally {
@@ -111,11 +114,14 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 	}
 
 	private int insertRows(Object key, PersistentCollection<?> collection, SharedSessionContractImplementor session) {
-		final MutationOperationGroupSingle operationGroup = resolveInsertGroup();
-
 		final PluralAttributeMapping attributeMapping = getMutationTarget().getTargetPart();
 		final CollectionPersister collectionDescriptor = attributeMapping.getCollectionDescriptor();
+		final Iterator<?> entries = collection.entries( collectionDescriptor );
+		if ( !entries.hasNext() ) {
+			return -1;
+		}
 
+		final MutationOperationGroupSingle operationGroup = resolveInsertGroup();
 		final MutationExecutorService mutationExecutorService = session
 				.getFactory()
 				.getServiceRegistry()
@@ -129,7 +135,6 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 		try {
 			final JdbcValueBindings jdbcValueBindings = mutationExecutor.getJdbcValueBindings();
 
-			final Iterator<?> entries = collection.entries( collectionDescriptor );
 			int entryPosition = -1;
 
 			while ( entries.hasNext() ) {

hibernate-core/src/main/java/org/hibernate/persister/collection/mutation/UpdateRowsCoordinatorTablePerSubclass.java

@@ -0,0 +1,257 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or http://www.gnu.org/licenses/lgpl-2.1.html.
+ */
+package org.hibernate.persister.collection.mutation;
+
+import java.util.Iterator;
+
+import org.hibernate.collection.spi.PersistentCollection;
+import org.hibernate.engine.jdbc.batch.internal.BasicBatchKey;
+import org.hibernate.engine.jdbc.mutation.MutationExecutor;
+import org.hibernate.engine.jdbc.mutation.spi.BatchKeyAccess;
+import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
+import org.hibernate.engine.spi.EntityEntry;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.PluralAttributeMapping;
+import org.hibernate.persister.collection.CollectionPersister;
+import org.hibernate.persister.collection.OneToManyPersister;
+import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.sql.model.MutationType;
+import org.hibernate.sql.model.internal.MutationOperationGroupSingle;
+import org.hibernate.sql.model.jdbc.JdbcMutationOperation;
+
+/**
+ * OneToMany update coordinator if the element is a {@link org.hibernate.persister.entity.UnionSubclassEntityPersister}.
+ */
+public class UpdateRowsCoordinatorTablePerSubclass extends AbstractUpdateRowsCoordinator {
+	private final RowMutationOperations rowMutationOperations;
+
+	private final SubclassEntry[] deleteSubclassEntries;
+	private final SubclassEntry[] insertSubclassEntries;
+
+	public UpdateRowsCoordinatorTablePerSubclass(
+			OneToManyPersister mutationTarget,
+			RowMutationOperations rowMutationOperations,
+			SessionFactoryImplementor sessionFactory) {
+		super( mutationTarget, sessionFactory );
+		this.rowMutationOperations = rowMutationOperations;
+		this.deleteSubclassEntries = new SubclassEntry[mutationTarget.getElementPersister().getRootEntityDescriptor().getSubclassEntityNames().size()];
+		this.insertSubclassEntries = new SubclassEntry[mutationTarget.getElementPersister().getRootEntityDescriptor().getSubclassEntityNames().size()];
+	}
+
+	@Override
+	protected int doUpdate(Object key, PersistentCollection<?> collection, SharedSessionContractImplementor session) {
+		if ( rowMutationOperations.hasDeleteRow() ) {
+			deleteRows( key, collection, session );
+		}
+
+		if ( rowMutationOperations.hasInsertRow() ) {
+			return insertRows( key, collection, session );
+		}
+
+		return 0;
+	}
+
+	private void deleteRows(Object key, PersistentCollection<?> collection, SharedSessionContractImplementor session) {
+		final PluralAttributeMapping attributeMapping = getMutationTarget().getTargetPart();
+		final CollectionPersister collectionDescriptor = attributeMapping.getCollectionDescriptor();
+		final Iterator<?> entries = collection.entries( collectionDescriptor );
+		if ( !entries.hasNext() ) {
+			return;
+		}
+
+		final MutationExecutorService mutationExecutorService = session
+				.getFactory()
+				.getServiceRegistry()
+				.getService( MutationExecutorService.class );
+
+		final MutationExecutor[] executors = new MutationExecutor[deleteSubclassEntries.length];
+		try {
+			int entryPosition = -1;
+
+			while ( entries.hasNext() ) {
+				final Object entry = entries.next();
+				entryPosition++;
+
+				if ( !collection.needsUpdating( entry, entryPosition, attributeMapping ) ) {
+					continue;
+				}
+				final Object entryToUpdate = collection.getSnapshotElement( entry, entryPosition );
+
+				final EntityEntry entityEntry = session.getPersistenceContextInternal().getEntry( entryToUpdate );
+				final int subclassId = entityEntry.getPersister().getSubclassId();
+				final MutationExecutor mutationExecutor;
+				if ( executors[subclassId] == null ) {
+					final SubclassEntry subclassEntry = getDeleteSubclassEntry( entityEntry.getPersister() );
+					mutationExecutor = executors[subclassId] = mutationExecutorService.createExecutor(
+							subclassEntry.batchKeySupplier,
+							subclassEntry.operationGroup,
+							session
+					);
+				}
+				else {
+					mutationExecutor = executors[subclassId];
+				}
+				rowMutationOperations.getDeleteRowRestrictions().applyRestrictions(
+						collection,
+						key,
+						entryToUpdate,
+						entryPosition,
+						session,
+						mutationExecutor.getJdbcValueBindings()
+				);
+
+				mutationExecutor.execute( entryToUpdate, null, null, null, session );
+			}
+		}
+		finally {
+			for ( MutationExecutor executor : executors ) {
+				if ( executor != null ) {
+					executor.release();
+				}
+			}
+		}
+	}
+
+	private SubclassEntry getDeleteSubclassEntry( EntityPersister elementPersister) {
+		final int subclassId = elementPersister.getSubclassId();
+		final SubclassEntry subclassEntry = deleteSubclassEntries[subclassId];
+		if ( subclassEntry != null ) {
+			return subclassEntry;
+		}
+		final BasicBatchKey basicBatchKey = new BasicBatchKey( getMutationTarget().getRolePath() + "#UPDATE-DELETE#" + subclassId );
+		return deleteSubclassEntries[subclassId] = new SubclassEntry(
+				() -> basicBatchKey,
+				resolveDeleteGroup( elementPersister )
+		);
+	}
+
+	private MutationOperationGroupSingle resolveDeleteGroup(EntityPersister elementPersister) {
+		final CollectionTableMapping collectionTableMapping = getMutationTarget().getCollectionTableMapping();
+		final JdbcMutationOperation operation = rowMutationOperations.getDeleteRowOperation(
+				new CollectionTableMapping(
+						elementPersister.getMappedTableDetails().getTableName(),
+						collectionTableMapping.getSpaces(),
+						collectionTableMapping.isJoinTable(),
+						collectionTableMapping.isInverse(),
+						collectionTableMapping.getInsertDetails(),
+						collectionTableMapping.getUpdateDetails(),
+						collectionTableMapping.isCascadeDeleteEnabled(),
+						collectionTableMapping.getDeleteDetails(),
+						collectionTableMapping.getDeleteRowDetails()
+				)
+		);
+
+		return new MutationOperationGroupSingle( MutationType.DELETE, getMutationTarget(), operation );
+	}
+
+	private int insertRows(Object key, PersistentCollection<?> collection, SharedSessionContractImplementor session) {
+		final PluralAttributeMapping attributeMapping = getMutationTarget().getTargetPart();
+		final CollectionPersister collectionDescriptor = attributeMapping.getCollectionDescriptor();
+		final Iterator<?> entries = collection.entries( collectionDescriptor );
+		if ( !entries.hasNext() ) {
+			return -1;
+		}
+
+		final MutationExecutorService mutationExecutorService = session
+				.getFactory()
+				.getServiceRegistry()
+				.getService( MutationExecutorService.class );
+
+		final MutationExecutor[] executors = new MutationExecutor[insertSubclassEntries.length];
+		try {
+			int entryPosition = -1;
+
+			while ( entries.hasNext() ) {
+				final Object entry = entries.next();
+				entryPosition++;
+
+				if ( !collection.needsUpdating( entry, entryPosition, attributeMapping ) ) {
+					continue;
+				}
+
+				final EntityEntry entityEntry = session.getPersistenceContextInternal().getEntry( entry );
+				final int subclassId = entityEntry.getPersister().getSubclassId();
+				final MutationExecutor mutationExecutor;
+				if ( executors[subclassId] == null ) {
+					final SubclassEntry subclassEntry = getInsertSubclassEntry( entityEntry.getPersister() );
+					mutationExecutor = executors[subclassId] = mutationExecutorService.createExecutor(
+							subclassEntry.batchKeySupplier,
+							subclassEntry.operationGroup,
+							session
+					);
+				}
+				else {
+					mutationExecutor = executors[subclassId];
+				}
+				rowMutationOperations.getInsertRowValues().applyValues(
+						collection,
+						key,
+						entry,
+						entryPosition,
+						session,
+						mutationExecutor.getJdbcValueBindings()
+				);
+
+				mutationExecutor.execute( entry, null, null, null, session );
+			}
+
+			return entryPosition;
+		}
+		finally {
+			for ( MutationExecutor executor : executors ) {
+				if ( executor != null ) {
+					executor.release();
+				}
+			}
+		}
+	}
+
+	private SubclassEntry getInsertSubclassEntry( EntityPersister elementPersister) {
+		final int subclassId = elementPersister.getSubclassId();
+		final SubclassEntry subclassEntry = insertSubclassEntries[subclassId];
+		if ( subclassEntry != null ) {
+			return subclassEntry;
+		}
+		final BasicBatchKey basicBatchKey = new BasicBatchKey( getMutationTarget().getRolePath() + "#UPDATE-INSERT#" + subclassId );
+		return insertSubclassEntries[subclassId] = new SubclassEntry(
+				() -> basicBatchKey,
+				resolveInsertGroup( elementPersister )
+		);
+	}
+
+	private MutationOperationGroupSingle resolveInsertGroup(EntityPersister elementPersister) {
+		final CollectionTableMapping collectionTableMapping = getMutationTarget().getCollectionTableMapping();
+		final JdbcMutationOperation operation = rowMutationOperations.getInsertRowOperation(
+				new CollectionTableMapping(
+						elementPersister.getMappedTableDetails().getTableName(),
+						collectionTableMapping.getSpaces(),
+						collectionTableMapping.isJoinTable(),
+						collectionTableMapping.isInverse(),
+						collectionTableMapping.getInsertDetails(),
+						collectionTableMapping.getUpdateDetails(),
+						collectionTableMapping.isCascadeDeleteEnabled(),
+						collectionTableMapping.getDeleteDetails(),
+						collectionTableMapping.getDeleteRowDetails()
+				)
+		);
+
+		return new MutationOperationGroupSingle( MutationType.INSERT, getMutationTarget(), operation );
+	}
+
+	private static class SubclassEntry {
+
+		private final BatchKeyAccess batchKeySupplier;
+
+		private final MutationOperationGroupSingle operationGroup;
+
+		public SubclassEntry(BatchKeyAccess batchKeySupplier, MutationOperationGroupSingle operationGroup) {
+			this.batchKeySupplier = batchKeySupplier;
+			this.operationGroup = operationGroup;
+		}
+	}
+}

hibernate-core/src/main/java/org/hibernate/sql/model/TableMapping.java

@@ -23,6 +23,10 @@ public interface TableMapping extends TableDetails {
 	 */
 	String getTableName();
 
+	default boolean containsTableName(String tableName) {
+		return getTableName().equals( tableName );
+	}
+
 	/**
 	 * The position of the table relative to others for the {@link MutationTarget}
 	 */

hibernate-core/src/test/java/org/hibernate/orm/test/inheritance/ManyToManyAbstractTablePerClassTest.java

@@ -0,0 +1,151 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or <http://www.gnu.org/licenses/lgpl-2.1.html>.
+ */
+package org.hibernate.orm.test.inheritance;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.hibernate.testing.TestForIssue;
+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.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Inheritance;
+import jakarta.persistence.InheritanceType;
+import jakarta.persistence.JoinTable;
+import jakarta.persistence.ManyToMany;
+import jakarta.persistence.OrderColumn;
+import jakarta.persistence.Table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@TestForIssue(jiraKey = "HHH-16358")
+@DomainModel(
+		annotatedClasses = {
+				ManyToManyAbstractTablePerClassTest.TablePerClassBase.class,
+				ManyToManyAbstractTablePerClassTest.TablePerClassSub1.class,
+				ManyToManyAbstractTablePerClassTest.TablePerClassSub2.class
+		}
+)
+@SessionFactory
+public class ManyToManyAbstractTablePerClassTest {
+
+	@Test
+	public void testAddAndRemove(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			final TablePerClassSub1 o1 = session.find( TablePerClassSub1.class, 1 );
+			assertNotNull( o1 );
+			assertEquals( 1, o1.childrenSet.size() );
+			assertEquals( 1, o1.childrenList.size() );
+			assertEquals( 1, o1.childrenMap.size() );
+			TablePerClassBase o2 = o1.childrenSet.iterator().next();
+			assertEquals( 2, o2.id );
+			assertEquals( 2, o1.childrenList.get( 0 ).id );
+			assertEquals( 2, o1.childrenMap.get( 2 ).id );
+			o1.childrenSet.remove( o2 );
+			o1.childrenList.remove( 0 );
+			o1.childrenMap.remove( 2 );
+			TablePerClassSub1 o3 = new TablePerClassSub1( 3 );
+			session.persist( o3 );
+			o1.childrenSet.add( o3 );
+			o1.childrenList.add( o3 );
+			o1.childrenMap.put( 3, o3 );
+			session.flush();
+		} );
+		scope.inTransaction( session -> {
+			final TablePerClassSub1 o1 = session.find( TablePerClassSub1.class, 1 );
+			assertNotNull( o1 );
+			assertEquals( 1, o1.childrenSet.size() );
+			assertEquals( 1, o1.childrenList.size() );
+			assertEquals( 1, o1.childrenMap.size() );
+			TablePerClassBase o2 = o1.childrenSet.iterator().next();
+			assertEquals( 3, o2.id );
+			assertEquals( 3, o1.childrenList.get( 0 ).id );
+			assertEquals( 3, o1.childrenMap.get( 3 ).id );
+		} );
+	}
+
+	@BeforeEach
+	public void setupData(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			TablePerClassSub1 o1 = new TablePerClassSub1( 1 );
+			TablePerClassSub2 o2 = new TablePerClassSub2( 2 );
+			o1.childrenSet.add( o2 );
+			o1.childrenList.add( o2 );
+
+			session.persist( o2 );
+			session.persist( o1 );
+
+			o1.childrenMap.put( 2, o2 );
+		} );
+	}
+
+	@AfterEach
+	public void cleanupData(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			session.createQuery( "from TablePerClassBase", TablePerClassBase.class )
+					.getResultList()
+					.forEach( session::remove );
+		} );
+	}
+
+	@Entity(name = "TablePerClassBase")
+	@Inheritance(strategy = InheritanceType.TABLE_PER_CLASS)
+	public static abstract class TablePerClassBase {
+		@Id
+		Integer id;
+		@ManyToMany
+		@JoinTable(name = "children_set")
+		Set<TablePerClassBase> childrenSet = new HashSet<>();
+		@ManyToMany
+		@JoinTable(name = "children_list")
+		@OrderColumn(name = "listIndex")
+		List<TablePerClassBase> childrenList = new ArrayList<>();
+		@ManyToMany
+		@JoinTable(name = "children_map")
+		Map<Integer, TablePerClassBase> childrenMap = new HashMap<>();
+
+		public TablePerClassBase() {
+		}
+
+		public TablePerClassBase(Integer id) {
+			this.id = id;
+		}
+	}
+
+	@Entity(name = "TablePerClassSub1")
+	@Table(name = "table_per_class_sub_1")
+	public static class TablePerClassSub1 extends TablePerClassBase {
+		public TablePerClassSub1() {
+		}
+
+		public TablePerClassSub1(Integer id) {
+			super( id );
+		}
+	}
+
+	@Entity(name = "TablePerClassSub2")
+	@Table(name = "table_per_class_sub_2")
+	public static class TablePerClassSub2 extends TablePerClassBase {
+		public TablePerClassSub2() {
+		}
+
+		public TablePerClassSub2(Integer id) {
+			super( id );
+		}
+	}
+}

hibernate-core/src/test/java/org/hibernate/orm/test/inheritance/OneToManyAbstractTablePerClassTest.java

@@ -0,0 +1,159 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or <http://www.gnu.org/licenses/lgpl-2.1.html>.
+ */
+package org.hibernate.orm.test.inheritance;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.hibernate.testing.TestForIssue;
+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.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import jakarta.persistence.CascadeType;
+import jakarta.persistence.Entity;
+import jakarta.persistence.FetchType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Inheritance;
+import jakarta.persistence.InheritanceType;
+import jakarta.persistence.JoinColumn;
+import jakarta.persistence.ManyToOne;
+import jakarta.persistence.MapKey;
+import jakarta.persistence.OneToMany;
+import jakarta.persistence.OrderColumn;
+import jakarta.persistence.Table;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+@TestForIssue(jiraKey = "HHH-16358")
+@DomainModel(
+		annotatedClasses = {
+				OneToManyAbstractTablePerClassTest.TablePerClassBase.class,
+				OneToManyAbstractTablePerClassTest.TablePerClassSub1.class,
+				OneToManyAbstractTablePerClassTest.TablePerClassSub2.class
+		}
+)
+@SessionFactory
+public class OneToManyAbstractTablePerClassTest {
+
+	@Test
+	public void testAddAndRemove(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			final TablePerClassSub1 o1 = session.find( TablePerClassSub1.class, 1 );
+			assertNotNull( o1 );
+			assertEquals( 1, o1.childrenSet.size() );
+			assertEquals( 1, o1.childrenList.size() );
+			assertEquals( 1, o1.childrenMap.size() );
+			TablePerClassBase o2 = o1.childrenSet.iterator().next();
+			assertEquals( 2, o2.id );
+			assertEquals( 2, o1.childrenList.get( 0 ).id );
+			assertEquals( 2, o1.childrenMap.get( 2 ).id );
+			o1.childrenSet.remove( o2 );
+			o1.childrenList.remove( 0 );
+			o1.childrenMap.remove( 2 );
+			o2.parent = null;
+			TablePerClassSub1 o3 = new TablePerClassSub1( 3 );
+			o3.parent = o1;
+			session.persist( o3 );
+			o1.childrenSet.add( o3 );
+			o1.childrenList.add( o3 );
+			o1.childrenMap.put( 3, o3 );
+			session.flush();
+		} );
+		scope.inTransaction( session -> {
+			final TablePerClassSub1 o1 = session.find( TablePerClassSub1.class, 1 );
+			assertNotNull( o1 );
+			assertEquals( 1, o1.childrenSet.size() );
+			assertEquals( 1, o1.childrenList.size() );
+			assertEquals( 1, o1.childrenMap.size() );
+			TablePerClassBase o2 = o1.childrenSet.iterator().next();
+			assertEquals( 3, o2.id );
+			assertEquals( 3, o1.childrenList.get( 0 ).id );
+			assertEquals( 3, o1.childrenMap.get( 3 ).id );
+		});
+	}
+
+	@BeforeEach
+	public void setupData(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			TablePerClassSub1 o1 = new TablePerClassSub1( 1 );
+			TablePerClassSub2 o2 = new TablePerClassSub2( 2 );
+			o1.childrenSet.add( o2 );
+			o1.childrenList.add( o2 );
+
+			session.persist( o2 );
+			session.persist( o1 );
+
+			o1.childrenMap.put( 2, o2 );
+			o2.parent = o1;
+		} );
+	}
+
+	@AfterEach
+	public void cleanupData(SessionFactoryScope scope) {
+		scope.inTransaction( session -> {
+			session.createQuery( "from TablePerClassBase", TablePerClassBase.class )
+					.getResultList()
+					.forEach( session::remove );
+		} );
+	}
+
+	@Entity(name = "TablePerClassBase")
+	@Inheritance(strategy = InheritanceType.TABLE_PER_CLASS)
+	public static abstract class TablePerClassBase {
+		@Id
+		Integer id;
+		@ManyToOne(fetch = FetchType.LAZY)
+		TablePerClassBase parent;
+		@OneToMany
+		@JoinColumn(name = "setParent")
+		Set<TablePerClassBase> childrenSet = new HashSet<>();
+		@OneToMany
+		@OrderColumn(name = "listIndex")
+		@JoinColumn(name = "listParent")
+		List<TablePerClassBase> childrenList = new ArrayList<>();
+		@MapKey(name = "id")
+		@OneToMany(mappedBy = "parent")
+		Map<Integer, TablePerClassBase> childrenMap = new HashMap<>();
+
+		public TablePerClassBase() {
+		}
+
+		public TablePerClassBase(Integer id) {
+			this.id = id;
+		}
+	}
+
+	@Entity(name = "TablePerClassSub1")
+	@Table(name = "table_per_class_sub_1")
+	public static class TablePerClassSub1 extends TablePerClassBase {
+		public TablePerClassSub1() {
+		}
+
+		public TablePerClassSub1(Integer id) {
+			super( id );
+		}
+	}
+	@Entity(name = "TablePerClassSub2")
+	@Table(name = "table_per_class_sub_2")
+	public static class TablePerClassSub2 extends TablePerClassBase {
+		public TablePerClassSub2() {
+		}
+
+		public TablePerClassSub2(Integer id) {
+			super( id );
+		}
+	}
+}

hibernate-core/src/test/java/org/hibernate/orm/test/jdbc/internal/BatchingTest.java

@@ -118,7 +118,12 @@ public class BatchingTest extends BaseCoreFunctionalTestCase implements BatchKey
 		return new JdbcValueBindingsImpl(
 				MutationType.INSERT,
 				null,
-				(tableName, columnName, usage) -> {
+				new JdbcValueBindingsImpl.JdbcValueDescriptorAccess() {
+					@Override
+					public JdbcValueDescriptor resolveValueDescriptor(
+							String tableName,
+							String columnName,
+							ParameterUsage usage) {
 						assert tableName.equals( SANDBOX_TBL );
 
 						if ( columnName.equals( "ID" ) ) {
@@ -174,6 +179,7 @@ public class BatchingTest extends BaseCoreFunctionalTestCase implements BatchKey
 						}
 
 						throw new IllegalArgumentException( "Unknown column : " + columnName );
+					}
 				},
 				session
 		);