Get rid of most capturing lambdas in write path

hibernate-core/src/main/java/org/hibernate/loader/ast/internal/CompoundNaturalIdLoader.java

@@ -71,7 +71,7 @@ public class CompoundNaturalIdLoader<T> extends AbstractNaturalIdLoader<T> {
 
 		naturalIdMapping().breakDownJdbcValues(
 				bindValue,
-				(jdbcValue, jdbcValueMapping) -> {
+				(valueIndex, jdbcValue, jdbcValueMapping) -> {
 					final Expression columnReference = resolveColumnReference(
 							rootTableGroup,
 							jdbcValueMapping,

hibernate-core/src/main/java/org/hibernate/loader/ast/internal/SimpleNaturalIdLoader.java

@@ -63,7 +63,7 @@ public class SimpleNaturalIdLoader<T> extends AbstractNaturalIdLoader<T> {
 		else {
 			naturalIdMapping().getAttribute().breakDownJdbcValues(
 					bindValue,
-					(jdbcValue, jdbcValueMapping) -> {
+					(valueIndex, jdbcValue, jdbcValueMapping) -> {
 						final Expression columnReference = resolveColumnReference(
 								rootTableGroup,
 								jdbcValueMapping,

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

@@ -218,14 +218,14 @@ public interface Bindable extends JdbcMappingContainer {
 	@FunctionalInterface
 	interface JdbcValuesConsumer extends JdbcValuesBiConsumer<Object, Object> {
 		@Override
-		default void consume(int selectionIndex, Object o, Object o2, Object jdbcValue, JdbcMapping jdbcMapping) {
+		default void consume(int valueIndex, Object o, Object o2, Object jdbcValue, JdbcMapping jdbcMapping) {
-			consume( selectionIndex, jdbcValue, jdbcMapping );
+			consume( valueIndex, jdbcValue, jdbcMapping );
 		}
 
 		/**
 		 * Consume a JDBC-level jdbcValue.  The JDBC jdbcMapping descriptor is also passed in
 		 */
-		void consume(int selectionIndex, Object jdbcValue, JdbcMapping jdbcMapping);
+		void consume(int valueIndex, Object jdbcValue, JdbcMapping jdbcMapping);
 	}
 
 	/**
@@ -236,6 +236,6 @@ public interface Bindable extends JdbcMappingContainer {
 		/**
 		 * Consume a JDBC-level jdbcValue.  The JDBC jdbcMapping descriptor is also passed in
 		 */
-		void consume(int selectionIndex, X x, Y y, Object jdbcValue, JdbcMapping jdbcMapping);
+		void consume(int valueIndex, X x, Y y, Object jdbcValue, JdbcMapping jdbcMapping);
 	}
 }

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

@@ -123,10 +123,17 @@ public interface ModelPart extends MappingModelExpressible {
 			DomainResultCreationState creationState,
 			BiConsumer<SqlSelection,JdbcMapping> selectionConsumer);
 
+	/**
+	 * A short hand form of {@link #forEachSelectable(int, SelectableConsumer)}, that passes 0 as offset.
+	 */
 	default int forEachSelectable(SelectableConsumer consumer) {
 		return forEachSelectable( 0, consumer );
 	}
 
+	/**
+	 * Visits each selectable mapping with the selectable index offset by the given value.
+	 * Returns the amount of jdbc types that have been visited.
+	 */
 	default int forEachSelectable(int offset, SelectableConsumer consumer) {
 		return 0;
 	}
@@ -135,21 +142,56 @@ public interface ModelPart extends MappingModelExpressible {
 		return null;
 	}
 
-	@FunctionalInterface
+	/**
-	interface JdbcValueConsumer {
+	 * A short hand form of {@link #breakDownJdbcValues(Object, int, Object, Object, JdbcValueBiConsumer, SharedSessionContractImplementor)},
-		void consume(Object value, SelectableMapping jdbcValueMapping);
+	 * that passes 0 as offset and null for the two values {@code X} and {@code Y}.
-
+	 */
+	default int breakDownJdbcValues(
+			Object domainValue,
+			JdbcValueConsumer valueConsumer,
+			SharedSessionContractImplementor session) {
+		return breakDownJdbcValues( domainValue, 0, null, null, valueConsumer, session );
 	}
 
-	void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session);
+	/**
+	 * Breaks down the domain value to its constituent JDBC values.
+	 *
+	 * Think of it as breaking the multi-dimensional array into a visitable flat array.
+	 * Additionally, it passes through the values {@code X} and {@code Y} to the consumer.
+	 * Returns the amount of jdbc types that have been visited.
+	 */
+	<X, Y> int breakDownJdbcValues(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session);
 
-	@FunctionalInterface
+	/**
-	interface IndexedJdbcValueConsumer {
+	 * A short hand form of {@link #decompose(Object, int, Object, Object, JdbcValueBiConsumer, SharedSessionContractImplementor)},
-		void consume(int valueIndex, Object value, SelectableMapping jdbcValueMapping);
+	 * that passes 0 as offset and null for the two values {@code X} and {@code Y}.
+	 */
+	default int decompose(
+			Object domainValue,
+			JdbcValueConsumer valueConsumer,
+			SharedSessionContractImplementor session) {
+		return decompose( domainValue, 0, null, null, valueConsumer, session );
 	}
 
-	default void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	/**
-		breakDownJdbcValues( domainValue, valueConsumer, session );
+	 * Similar to {@link #breakDownJdbcValues(Object, int, Object, Object, JdbcValueBiConsumer, SharedSessionContractImplementor)},
+	 * but this method is supposed to be used for decomposing values for assignment expressions.
+	 * Returns the amount of jdbc types that have been visited.
+	 */
+	default <X, Y> int decompose(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	EntityMappingType findContainingEntityMapping();
@@ -158,4 +200,31 @@ public interface ModelPart extends MappingModelExpressible {
 		// NOTE : deepEquals to account for arrays (compound natural-id)
 		return Objects.deepEquals( one, other );
 	}
+
+	/**
+	 * Functional interface for consuming the JDBC values.
+	 */
+	@FunctionalInterface
+	interface JdbcValueConsumer extends JdbcValueBiConsumer<Object, Object> {
+		@Override
+		default void consume(int valueIndex, Object x, Object y, Object value, SelectableMapping jdbcValueMapping) {
+			consume( valueIndex, value, jdbcValueMapping );
+		}
+
+		/**
+		 * Consume a JDBC-level jdbcValue.  The JDBC jdbcMapping descriptor is also passed in
+		 */
+		void consume(int valueIndex, Object value, SelectableMapping jdbcValueMapping);
+	}
+
+	/**
+	 * Functional interface for consuming the JDBC values, along with two values of type {@code X} and {@code Y}.
+	 */
+	@FunctionalInterface
+	interface JdbcValueBiConsumer<X, Y> {
+		/**
+		 * Consume a JDBC-level jdbcValue.  The JDBC jdbcMapping descriptor is also passed in
+		 */
+		void consume(int valueIndex, X x, Y y, Object value, SelectableMapping jdbcValueMapping);
+	}
 }

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

@@ -239,8 +239,15 @@ public abstract class AbstractDiscriminatorMapping implements EntityDiscriminato
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( disassemble( domainValue, session ), this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, disassemble( domainValue, session ), this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -225,8 +225,15 @@ public class AnyDiscriminatorPart implements BasicValuedModelPart, FetchOptions,
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -259,8 +259,15 @@ public class AnyKeyPart implements BasicValuedModelPart, FetchOptions {
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -423,7 +423,14 @@ public class BasicAttributeMapping
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( disassemble( domainValue, session ), this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, disassemble( domainValue, session ), this );
+		return getJdbcTypeCount();
 	}
 }

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

@@ -186,8 +186,15 @@ public class BasicEntityIdentifierMappingImpl implements BasicEntityIdentifierMa
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -324,13 +324,26 @@ public class BasicValuedCollectionPart
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( disassemble( domainValue, session ), this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, disassemble( domainValue, session ), this );
+		return getJdbcTypeCount();
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
-		valueConsumer.consume( disassemble( domainValue, session ), this );
+			Object domainValue, int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, disassemble( domainValue, session ), this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -198,12 +198,19 @@ public class CollectionIdentifierDescriptorImpl implements CollectionIdentifierD
 	@Override
 	public int forEachSelectable(int offset, SelectableConsumer consumer) {
 		consumer.accept( offset, this );
-		return 1;
+		return getJdbcTypeCount();
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -313,12 +313,19 @@ public class CompoundNaturalIdMapping extends AbstractNaturalIdMapping implement
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		int span = 0;
 		if ( domainValue == null ) {
-			attributes.forEach(
+			for ( int i = 0; i < attributes.size(); i++ ) {
-					attributeMapping -> attributeMapping.breakDownJdbcValues( null, valueConsumer, session )
+				span += attributes.get( i ).breakDownJdbcValues( null, offset + span, x, y, valueConsumer, session );
-			);
+			}
-			return;
+			return span;
 		}
 
 		assert domainValue instanceof Object[];
@@ -327,8 +334,9 @@ public class CompoundNaturalIdMapping extends AbstractNaturalIdMapping implement
 		assert values.length == attributes.size();
 
 		for ( int i = 0; i < attributes.size(); i++ ) {
-			attributes.get( i ).breakDownJdbcValues( values[ i ], valueConsumer, session );
+			span += attributes.get( i ).breakDownJdbcValues( values[ i ], offset + span, x, y, valueConsumer, session );
 		}
+		return span;
 	}
 
 	@Override

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

@@ -323,13 +323,25 @@ public class DiscriminatedAssociationAttributeMapping
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		discriminatorMapping.breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return discriminatorMapping.breakDownJdbcValues( offset, x, y, domainValue, valueConsumer, session );
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
-		discriminatorMapping.decompose( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return discriminatorMapping.decompose( offset, x, y, domainValue, valueConsumer, session );
 	}
 
 	@Override

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

@@ -206,43 +206,55 @@ public class DiscriminatedAssociationMapping implements MappingType, FetchOption
 		return null;
 	}
 
-	public void breakDownJdbcValues(Object domainValue, ModelPart.JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		if ( domainValue == null ) {
+			int offset,
-			valueConsumer.consume( null, getDiscriminatorPart() );
+			X x,
-			valueConsumer.consume( null, getKeyPart() );
+			Y y,
-			return;
-		}
-
-		final EntityMappingType concreteMappingType = determineConcreteType( domainValue, session );
-
-		final Object discriminator = getModelPart().resolveDiscriminatorForEntityType( concreteMappingType );
-		final Object disassembledDiscriminator = getDiscriminatorPart().disassemble( discriminator, session );
-		valueConsumer.consume( disassembledDiscriminator, getDiscriminatorPart() );
-
-		final EntityIdentifierMapping identifierMapping = concreteMappingType.getIdentifierMapping();
-		final Object identifier = identifierMapping.getIdentifier( domainValue );
-		final Object disassembledKey = getKeyPart().disassemble( identifier, session );
-		valueConsumer.consume( disassembledKey, getKeyPart() );
-	}
-
-	public void decompose(
 			Object domainValue,
-			ModelPart.JdbcValueConsumer valueConsumer,
+			ModelPart.JdbcValueBiConsumer<X, Y> valueConsumer,
 			SharedSessionContractImplementor session) {
 		if ( domainValue == null ) {
-			valueConsumer.consume( null, getDiscriminatorPart() );
+			valueConsumer.consume( offset, x, y, null, getDiscriminatorPart() );
-			valueConsumer.consume( null, getKeyPart() );
+			valueConsumer.consume( offset + 1, x, y, null, getKeyPart() );
-			return;
+			return getDiscriminatorPart().getJdbcTypeCount() + getKeyPart().getJdbcTypeCount();
 		}
+		else {
+			final EntityMappingType concreteMappingType = determineConcreteType( domainValue, session );
 
-		final EntityMappingType concreteMappingType = determineConcreteType( domainValue, session );
+			final Object discriminator = getModelPart().resolveDiscriminatorForEntityType( concreteMappingType );
+			final Object disassembledDiscriminator = getDiscriminatorPart().disassemble( discriminator, session );
+			valueConsumer.consume( offset, x, y, disassembledDiscriminator, getDiscriminatorPart() );
 
-		final Object discriminator = getModelPart().resolveDiscriminatorForEntityType( concreteMappingType );
+			final EntityIdentifierMapping identifierMapping = concreteMappingType.getIdentifierMapping();
-		getDiscriminatorPart().decompose( discriminator, valueConsumer, session );
+			final Object identifier = identifierMapping.getIdentifier( domainValue );
+			final Object disassembledKey = getKeyPart().disassemble( identifier, session );
+			valueConsumer.consume( offset + 1, x, y, disassembledKey, getKeyPart() );
+		}
+		return getDiscriminatorPart().getJdbcTypeCount() + getKeyPart().getJdbcTypeCount();
+	}
 
-		final EntityIdentifierMapping identifierMapping = concreteMappingType.getIdentifierMapping();
+	public <X, Y> int decompose(
-		final Object identifier = identifierMapping.getIdentifier( domainValue );
+			int offset,
-		getKeyPart().decompose( identifier, valueConsumer, session );
+			X x,
+			Y y,
+			Object domainValue,
+			ModelPart.JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		if ( domainValue == null ) {
+			valueConsumer.consume( offset, x, y, null, getDiscriminatorPart() );
+			valueConsumer.consume( offset + 1, x, y, null, getKeyPart() );
+		}
+		else {
+			final EntityMappingType concreteMappingType = determineConcreteType( domainValue, session );
+
+			final Object discriminator = getModelPart().resolveDiscriminatorForEntityType( concreteMappingType );
+			getDiscriminatorPart().decompose( discriminator, offset, x, y, valueConsumer, session );
+
+			final EntityIdentifierMapping identifierMapping = concreteMappingType.getIdentifierMapping();
+			final Object identifier = identifierMapping.getIdentifier( domainValue );
+			getKeyPart().decompose( identifier, offset + 1, x, y, valueConsumer, session );
+		}
+		return getDiscriminatorPart().getJdbcTypeCount() + getKeyPart().getJdbcTypeCount();
 	}
 
 	private EntityMappingType determineConcreteType(Object entity, SharedSessionContractImplementor session) {

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

@@ -285,13 +285,25 @@ public class DiscriminatedCollectionPart implements DiscriminatedAssociationMode
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		discriminatorMapping.breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return discriminatorMapping.breakDownJdbcValues( offset, x, y, domainValue, valueConsumer, session );
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
-		discriminatorMapping.decompose( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return discriminatorMapping.decompose( offset, x, y, domainValue, valueConsumer, session );
 	}
 
 	@Override

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

@@ -695,8 +695,15 @@ public class EmbeddableMappingTypeImpl extends AbstractEmbeddableMapping impleme
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
 		final int size = attributeMappings.size();
+		int span = 0;
 		if ( domainValue instanceof Object[] ) {
 			final Object[] values = (Object[]) domainValue;
 			assert values.length == size;
@@ -704,7 +711,14 @@ public class EmbeddableMappingTypeImpl extends AbstractEmbeddableMapping impleme
 			for ( int i = 0; i < size; i++ ) {
 				final AttributeMapping attributeMapping = attributeMappings.get( i );
 				final Object attributeValue = values[ i ];
-				attributeMapping.breakDownJdbcValues( attributeValue, valueConsumer, session );
+				span += attributeMapping.breakDownJdbcValues(
+						attributeValue,
+						offset + span,
+						x,
+						y,
+						valueConsumer,
+						session
+				);
 			}
 		}
 		else {
@@ -713,38 +727,54 @@ public class EmbeddableMappingTypeImpl extends AbstractEmbeddableMapping impleme
 				final Object attributeValue = domainValue == null
 						? null
 						: attributeMapping.getPropertyAccess().getGetter().get( domainValue );
-				attributeMapping.breakDownJdbcValues( attributeValue, valueConsumer, session );
+				span += attributeMapping.breakDownJdbcValues(
+						attributeValue,
+						offset + span,
+						x,
+						y,
+						valueConsumer,
+						session
+				);
 			}
 		}
+		return span;
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
 		if ( shouldBindAggregateMapping() ) {
-			valueConsumer.consume( domainValue, aggregateMapping );
+			valueConsumer.consume( offset, x, y, domainValue, aggregateMapping );
+			return 1;
 		}
-		else if ( domainValue instanceof Object[] ) {
+		int span = 0;
+		if ( domainValue instanceof Object[] ) {
 			final Object[] values = (Object[]) domainValue;
 			assert values.length == attributeMappings.size();
 
 			for ( int i = 0; i < attributeMappings.size(); i++ ) {
 				final AttributeMapping attributeMapping = attributeMappings.get( i );
 				final Object attributeValue = values[ i ];
-				attributeMapping.decompose( attributeValue, valueConsumer, session );
+				span += attributeMapping.decompose( attributeValue, offset + span, x, y, valueConsumer, session );
 			}
 		}
 		else {
-			attributeMappings.forEach( (attributeMapping) -> {
+			for ( int i = 0; i < attributeMappings.size(); i++ ) {
-				if ( attributeMapping instanceof PluralAttributeMapping ) {
+				final AttributeMapping attributeMapping = attributeMappings.get( i );
-					return;
+				if ( !(attributeMapping instanceof PluralAttributeMapping )) {
+					final Object attributeValue = domainValue == null
+							? null
+							: attributeMapping.getPropertyAccess().getGetter().get( domainValue );
+					span += attributeMapping.decompose( attributeValue, offset + span, x, y, valueConsumer, session );
 				}
-
+			}
-				final Object attributeValue = domainValue == null
-						? null
-						: attributeMapping.getPropertyAccess().getGetter().get( domainValue );
-				attributeMapping.decompose( attributeValue, valueConsumer, session );
-			} );
 		}
+		return span;
 	}
 
 	@Override

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

@@ -208,13 +208,25 @@ public class EmbeddedAttributeMapping
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		getEmbeddableTypeDescriptor().breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return getEmbeddableTypeDescriptor().breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
-		getEmbeddableTypeDescriptor().decompose( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return getEmbeddableTypeDescriptor().decompose( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

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

@@ -341,8 +341,14 @@ public class EmbeddedCollectionPart implements CollectionPart, EmbeddableValuedF
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		getEmbeddableTypeDescriptor().breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return getEmbeddableTypeDescriptor().breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

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

@@ -532,23 +532,39 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor {
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
 		if ( domainValue == null ) {
-			keySelectableMappings.forEachSelectable( (index, selectable) -> {
+			final int jdbcTypeCount = keySelectableMappings.getJdbcTypeCount();
-				valueConsumer.consume( null, selectable );
+			for ( int i = 0; i < jdbcTypeCount; i++ ) {
-			} );
+				valueConsumer.consume( offset + i, x, y, null, keySelectableMappings.getSelectable( i ) );
+			}
+			return jdbcTypeCount;
 		}
 		else if ( domainValue instanceof Object[] ) {
 			final Object[] values = (Object[]) domainValue;
-			keySelectableMappings.forEachSelectable( (index, selectable) -> {
+			final int jdbcTypeCount = keySelectableMappings.getJdbcTypeCount();
-				valueConsumer.consume( values[ index ], selectable );
+			for ( int i = 0; i < jdbcTypeCount; i++ ) {
-			} );
+				valueConsumer.consume( offset + i, x, y, values[i], keySelectableMappings.getSelectable( i ) );
+			}
+			return jdbcTypeCount;
 		}
 		else {
 			final MutableInteger columnPosition = new MutableInteger();
-			keySide.getModelPart().breakDownJdbcValues(
+			return keySide.getModelPart().breakDownJdbcValues(
 					domainValue,
-					(jdbcValue, jdbcValueMapping) -> valueConsumer.consume(
+					offset,
+					x,
+					y,
+					(valueIndex, arg1, arg2, jdbcValue, jdbcValueMapping) -> valueConsumer.consume(
+							offset,
+							arg1,
+							arg2,
 							jdbcValue,
 							keySelectableMappings.getSelectable( columnPosition.getAndIncrement() )
 					),

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

@@ -131,8 +131,14 @@ public class EmbeddedIdentifierMappingImpl
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		getEmbeddableTypeDescriptor().breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return getEmbeddableTypeDescriptor().breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

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

@@ -154,8 +154,15 @@ public class EntityRowIdMappingImpl implements EntityRowIdMapping {
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -290,8 +290,15 @@ public class EntityVersionMappingImpl implements EntityVersionMapping, FetchOpti
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	private SqlSelection resolveSqlSelection(TableGroup tableGroup, DomainResultCreationState creationState) {

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

@@ -367,11 +367,20 @@ public class IdClassEmbeddable extends AbstractEmbeddableMapping implements Iden
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		attributeMappings.forEach( (attribute) -> {
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		int span = 0;
+		for ( int i = 0; i < attributeMappings.size(); i++ ) {
+			final AttributeMapping attribute = attributeMappings.get( i );
 			final Object attributeValue = attribute.getValue( domainValue );
-			attribute.breakDownJdbcValues( attributeValue, valueConsumer, session );
+			span +=  attribute.breakDownJdbcValues( attributeValue, offset + span, x, y, valueConsumer, session );
-		} );
+		}
+		return span;
 	}
 
 	@Override

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

@@ -262,8 +262,13 @@ public class InverseNonAggregatedIdentifierMapping extends EmbeddedAttributeMapp
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		identifierValueMapper.breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer, SharedSessionContractImplementor session) {
+		return identifierValueMapper.breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

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

@@ -137,8 +137,13 @@ public class ManyToManyCollectionPart extends AbstractEntityCollectionPart imple
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		fkTargetModelPart.breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer, SharedSessionContractImplementor session) {
+		return fkTargetModelPart.breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override
@@ -158,9 +163,18 @@ public class ManyToManyCollectionPart extends AbstractEntityCollectionPart imple
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
-		foreignKey.getKeyPart().decompose(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return foreignKey.getKeyPart().decompose(
 				foreignKey.getAssociationKeyFromSide( domainValue, foreignKey.getTargetSide(), session ),
+				offset,
+				x,
+				y,
 				valueConsumer,
 				session
 		);

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

@@ -284,8 +284,13 @@ public class NonAggregatedIdentifierMappingImpl extends AbstractCompositeIdentif
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		identifierValueMapper.breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer, SharedSessionContractImplementor session) {
+		return identifierValueMapper.breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

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

@@ -82,12 +82,18 @@ public class OneToManyCollectionPart extends AbstractEntityCollectionPart implem
 	}
 
 	@Override
-	public void breakDownJdbcValues(
+	public <X, Y> int breakDownJdbcValues(
 			Object domainValue,
-			JdbcValueConsumer valueConsumer,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
 			SharedSessionContractImplementor session) {
-		getAssociatedEntityMappingType().getIdentifierMapping().breakDownJdbcValues(
+		return getAssociatedEntityMappingType().getIdentifierMapping().breakDownJdbcValues(
 				disassemble( domainValue, session ),
+				offset,
+				x,
+				y,
 				valueConsumer,
 				session
 		);

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

@@ -912,7 +912,13 @@ public class PluralAttributeMappingImpl
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
 		throw new UnsupportedOperationException();
 	}
 

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

@@ -462,12 +462,19 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 			JdbcValuesBiConsumer<X, Y> valuesConsumer,
 			SharedSessionContractImplementor session) {
 		valuesConsumer.consume( offset, x, y, value, getJdbcMapping() );
-		return 1;
+		return getJdbcTypeCount();
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( disassemble( domainValue, session ), keySide.getModelPart() );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, disassemble( domainValue, session ), keySide.getModelPart() );
+		return getJdbcTypeCount();
 	}
 
 	@Override

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

@@ -243,8 +243,14 @@ public class SimpleNaturalIdMapping extends AbstractNaturalIdMapping implements
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		attribute.breakDownJdbcValues( domainValue, valueConsumer, session );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		return attribute.breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

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

@@ -2022,16 +2022,19 @@ public class ToOneAttributeMapping
 	}
 
 	@Override
-	public void breakDownJdbcValues(
+	public <X, Y> int breakDownJdbcValues(
 			Object domainValue,
-			JdbcValueConsumer valueConsumer,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
 			SharedSessionContractImplementor session) {
 		if ( cardinality == Cardinality.ONE_TO_ONE && sideNature == ForeignKeyDescriptor.Nature.TARGET ) {
-			return;
+			return 0;
 		}
 
 		final Object value = extractValue( domainValue, session );
-		foreignKeyDescriptor.breakDownJdbcValues( value, valueConsumer, session );
+		return foreignKeyDescriptor.breakDownJdbcValues( value, offset, x, y, valueConsumer, session );
 	}
 
 	private Object extractValue(Object domainValue, SharedSessionContractImplementor session) {

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

@@ -319,29 +319,47 @@ public class VirtualIdEmbeddable extends AbstractEmbeddableMapping implements Id
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		attributeMappings.forEach( (attribute) -> {
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer, SharedSessionContractImplementor session) {
+		int span = 0;
+		for ( int i = 0; i < attributeMappings.size(); i++ ) {
+			final AttributeMapping attribute = attributeMappings.get( i );
 			final Object attributeValue = attribute.getValue( domainValue );
-			attribute.breakDownJdbcValues( attributeValue, valueConsumer, session );
+			span += attribute.breakDownJdbcValues( attributeValue, offset + span, x, y, valueConsumer, session );
-		} );
+		}
+		return span;
 	}
 
 	@Override
-	public void decompose(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int decompose(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer, SharedSessionContractImplementor session) {
 		if ( idMapping.getIdClassEmbeddable() != null ) {
 			// during decompose, if there is an IdClass for the entity the
 			// incoming `domainValue` should be an instance of that IdClass
-			idMapping.getIdClassEmbeddable().decompose( domainValue, valueConsumer, session );
+			return idMapping.getIdClassEmbeddable().decompose( domainValue, offset, x, y, valueConsumer, session );
 		}
 		else {
+			int span = 0;
 			for ( int i = 0; i < attributeMappings.size(); i++ ) {
 				final AttributeMapping attributeMapping = attributeMappings.get( i );
-				attributeMapping.decompose(
+				span += attributeMapping.decompose(
 						attributeMapping.getValue( domainValue ),
+						offset + span,
+						x,
+						y,
 						valueConsumer,
 						session
 				);
 			}
+			return span;
 		}
 	}
 

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

@@ -11,6 +11,7 @@ import org.hibernate.MappingException;
 import org.hibernate.cache.CacheException;
 import org.hibernate.cache.spi.access.CollectionDataAccess;
 import org.hibernate.collection.spi.PersistentCollection;
+import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
 import org.hibernate.engine.jdbc.mutation.ParameterUsage;
 import org.hibernate.engine.jdbc.mutation.internal.MutationQueryOptions;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -22,7 +23,6 @@ import org.hibernate.mapping.Collection;
 import org.hibernate.metamodel.mapping.CollectionIdentifierDescriptor;
 import org.hibernate.metamodel.mapping.CollectionPart;
 import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
-import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.metamodel.spi.RuntimeModelCreationContext;
 import org.hibernate.persister.collection.mutation.CollectionTableMapping;
@@ -331,7 +331,7 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 			Object rowValue,
 			int rowPosition,
 			SharedSessionContractImplementor session,
-			RowMutationOperations.ValuesBindingConsumer jdbcValueConsumer) {
+			JdbcValueBindings jdbcValueBindings) {
 		final PluralAttributeMapping attributeMapping = getAttributeMapping();
 
 		if ( key == null ) {
@@ -339,14 +339,24 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 		}
 
 		final ForeignKeyDescriptor foreignKey = attributeMapping.getKeyDescriptor();
-		foreignKey.getKeyPart().decompose( key, jdbcValueConsumer, session );
+		foreignKey.getKeyPart().decompose(
+				key,
+				0,
+				jdbcValueBindings,
+				null,
+				RowMutationOperations.DEFAULT_VALUE_SETTER,
+				session
+		);
 
 		final MutableInteger columnPositionCount = new MutableInteger();
 
 		if ( attributeMapping.getIdentifierDescriptor() != null ) {
 			getAttributeMapping().getIdentifierDescriptor().decompose(
 					collection.getIdentifier( rowValue, rowPosition ),
-					jdbcValueConsumer,
+					0,
+					jdbcValueBindings,
+					null,
+					RowMutationOperations.DEFAULT_VALUE_SETTER,
 					session
 			);
 		}
@@ -361,15 +371,17 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 
 			getAttributeMapping().getIndexDescriptor().decompose(
 					incrementIndexByBase( collection.getIndex( rowValue, rowPosition, this ) ),
-					(jdbcValue, jdbcValueMapping) -> {
+					0,
+					indexColumnIsSettable,
+					jdbcValueBindings,
+					(valueIndex, settable, bindings, jdbcValue, jdbcValueMapping) -> {
 						if ( !jdbcValueMapping.getContainingTableExpression().equals( getTableName() ) ) {
 							// indicates a many-to-many mapping and the index is contained on the
 							// associated entity table - we skip it here
 							return;
 						}
-						final int columnPosition = columnPositionCount.getAndIncrement();
+						if ( settable[valueIndex] ) {
-						if ( indexColumnIsSettable[columnPosition] ) {
+							bindings.bindValue( jdbcValue, jdbcValueMapping, ParameterUsage.SET );
-							jdbcValueConsumer.consume( jdbcValue, jdbcValueMapping );
 						}
 					},
 					session
@@ -380,10 +392,12 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 
 		attributeMapping.getElementDescriptor().decompose(
 				collection.getElement( rowValue ),
-				(jdbcValue, jdbcValueMapping) -> {
+				0,
-					final int columnPosition = columnPositionCount.getAndIncrement();
+				elementColumnIsSettable,
-					if ( elementColumnIsSettable[columnPosition] ) {
+				jdbcValueBindings,
-						jdbcValueConsumer.consume( jdbcValue, jdbcValueMapping );
+				(valueIndex, settable, bindings, jdbcValue, jdbcValueMapping) -> {
+					if ( settable[valueIndex] ) {
+						bindings.bindValue( jdbcValue, jdbcValueMapping, ParameterUsage.SET );
 					}
 				},
 				session
@@ -477,16 +491,19 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 			Object entry,
 			int entryPosition,
 			SharedSessionContractImplementor session,
-			RowMutationOperations.ValuesBindingConsumer jdbcValueConsumer) {
+			JdbcValueBindings jdbcValueBindings) {
 		final Object element = collection.getElement( entry );
 		final CollectionPart elementDescriptor = getAttributeMapping().getElementDescriptor();
 		elementDescriptor.decompose(
 				element,
-				(jdbcValue, jdbcValueMapping) -> {
+				0,
+				jdbcValueBindings,
+				null,
+				(valueIndex, bindings, y, jdbcValue, jdbcValueMapping) -> {
 					if ( !jdbcValueMapping.isUpdateable() || jdbcValueMapping.isFormula() ) {
 						return;
 					}
-					jdbcValueConsumer.consumeJdbcValueBinding(
+					bindings.bindValue(
 							jdbcValue,
 							jdbcValueMapping,
 							ParameterUsage.SET
@@ -502,29 +519,53 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 			Object entry,
 			int entryPosition,
 			SharedSessionContractImplementor session,
-			ModelPart.JdbcValueConsumer restrictor) {
+			JdbcValueBindings jdbcValueBindings) {
 		if ( getAttributeMapping().getIdentifierDescriptor() != null ) {
 			final CollectionIdentifierDescriptor identifierDescriptor = getAttributeMapping().getIdentifierDescriptor();
 			final Object identifier = collection.getIdentifier( entry, entryPosition );
-			identifierDescriptor.decompose( identifier, restrictor, session );
+			identifierDescriptor.decompose(
+					identifier,
+					0,
+					jdbcValueBindings,
+					null,
+					RowMutationOperations.DEFAULT_RESTRICTOR,
+					session
+			);
 		}
 		else {
-			getAttributeMapping().getKeyDescriptor().getKeyPart().decompose( key, restrictor, session );
+			getAttributeMapping().getKeyDescriptor().getKeyPart().decompose(
+					key,
+					0,
+					jdbcValueBindings,
+					null,
+					RowMutationOperations.DEFAULT_RESTRICTOR,
+					session
+			);
 
 			if ( getAttributeMapping().getIndexDescriptor() != null && !indexContainsFormula ) {
 				final Object index = collection.getIndex( entry, entryPosition, getAttributeMapping().getCollectionDescriptor() );
 				final Object adjustedIndex = incrementIndexByBase( index );
-				getAttributeMapping().getIndexDescriptor().decompose( adjustedIndex, restrictor, session );
+				getAttributeMapping().getIndexDescriptor().decompose(
+						adjustedIndex,
+						0,
+						jdbcValueBindings,
+						null,
+						RowMutationOperations.DEFAULT_RESTRICTOR,
+						session
+				);
 			}
 			else {
 				final Object snapshotElement = collection.getSnapshotElement( entry, entryPosition );
 				getAttributeMapping().getElementDescriptor().decompose(
 						snapshotElement,
-						(jdbcValue, jdbcValueMapping) -> {
+						0,
+						jdbcValueBindings,
+						null,
+						(valueIndex, bindings, noop, jdbcValue, jdbcValueMapping) -> {
 							if ( jdbcValueMapping.isNullable() || jdbcValueMapping.isFormula() ) {
 								return;
 							}
-							restrictor.consume( jdbcValue, jdbcValueMapping );
+							bindings.bindValue( jdbcValue, jdbcValueMapping, ParameterUsage.RESTRICT );
 						},
 						session
 				);
@@ -616,30 +657,50 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 			Object rowValue,
 			int rowPosition,
 			SharedSessionContractImplementor session,
-			ModelPart.JdbcValueConsumer restrictor) {
+			JdbcValueBindings jdbcValueBindings) {
 		final PluralAttributeMapping attributeMapping = getAttributeMapping();
 
 		if ( attributeMapping.getIdentifierDescriptor() != null ) {
-			attributeMapping.getIdentifierDescriptor().decompose( rowValue, restrictor, session );
+			attributeMapping.getIdentifierDescriptor().decompose(
+					rowValue,
+					0,
+					jdbcValueBindings,
+					null,
+					RowMutationOperations.DEFAULT_RESTRICTOR,
+					session
+			);
 		}
 		else {
-			getAttributeMapping().getKeyDescriptor().getKeyPart().decompose( keyValue, restrictor, session );
+			getAttributeMapping().getKeyDescriptor().getKeyPart().decompose(
+					keyValue,
+					0,
+					jdbcValueBindings,
+					null,
+					RowMutationOperations.DEFAULT_RESTRICTOR,
+					session
+			);
 
 			if ( hasPhysicalIndexColumn() ) {
 				attributeMapping.getIndexDescriptor().decompose(
 						incrementIndexByBase( rowValue ),
-						restrictor,
+						0,
+						jdbcValueBindings,
+						null,
+						RowMutationOperations.DEFAULT_RESTRICTOR,
 						session
 				);
 			}
 			else {
 				attributeMapping.getElementDescriptor().decompose(
 						rowValue,
-						(jdbcValue, jdbcValueMapping) -> {
+						0,
+						jdbcValueBindings,
+						null,
+						(valueIndex, bindings, noop, jdbcValue, jdbcValueMapping) -> {
 							if ( jdbcValueMapping.isNullable() || jdbcValueMapping.isFormula() ) {
 								return;
 							}
-							restrictor.consume( jdbcValue, jdbcValueMapping );
+							bindings.bindValue( jdbcValue, jdbcValueMapping, ParameterUsage.RESTRICT );
 						},
 						session
 				);

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

@@ -31,7 +31,6 @@ import org.hibernate.metamodel.mapping.CollectionPart;
 import org.hibernate.metamodel.mapping.EntityIdentifierMapping;
 import org.hibernate.metamodel.mapping.EntityMappingType;
 import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
-import org.hibernate.metamodel.mapping.ModelPart.JdbcValueConsumer;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
 import org.hibernate.metamodel.mapping.SelectableMapping;
 import org.hibernate.metamodel.mapping.internal.EntityCollectionPart;
@@ -243,11 +242,7 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 							entry,
 							nextIndex,
 							session,
-							(jdbcValue, jdbcValueMapping, usage) -> jdbcValueBindings.bindValue(
+							jdbcValueBindings
-									jdbcValue,
-									jdbcValueMapping,
-									usage
-							)
 					);
 
 					updateRowRestrictions.applyRestrictions(
@@ -256,11 +251,7 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 							entry,
 							nextIndex,
 							session,
-							(jdbcValue, jdbcValueMapping) -> jdbcValueBindings.bindValue(
+							jdbcValueBindings
-									jdbcValue,
-									jdbcValueMapping,
-									ParameterUsage.RESTRICT
-							)
 					);
 
 					mutationExecutor.execute( collection, null, null, null, session );
@@ -621,10 +612,24 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			Object rowValue,
 			int rowPosition,
 			SharedSessionContractImplementor session,
-			JdbcValueConsumer jdbcValueConsumer) {
+			JdbcValueBindings jdbcValueBindings) {
 		final PluralAttributeMapping pluralAttribute = getAttributeMapping();
-		pluralAttribute.getKeyDescriptor().decompose( keyValue, jdbcValueConsumer, session );
+		pluralAttribute.getKeyDescriptor().decompose(
-		pluralAttribute.getElementDescriptor().decompose( rowValue, jdbcValueConsumer, session );
+				keyValue,
+				0,
+				jdbcValueBindings,
+				null,
+				RowMutationOperations.DEFAULT_RESTRICTOR,
+				session
+		);
+		pluralAttribute.getElementDescriptor().decompose(
+				rowValue,
+				0,
+				jdbcValueBindings,
+				null,
+				RowMutationOperations.DEFAULT_RESTRICTOR,
+				session
+		);
 	}
 
 
@@ -658,21 +663,31 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			Object rowValue,
 			int rowPosition,
 			SharedSessionContractImplementor session,
-			RowMutationOperations.ValuesBindingConsumer bindingsConsumer) {
+			JdbcValueBindings jdbcValueBindings) {
 		final PluralAttributeMapping attributeMapping = getAttributeMapping();
 
-		attributeMapping.getKeyDescriptor().getKeyPart().decompose( keyValue, bindingsConsumer, session );
+		attributeMapping.getKeyDescriptor().getKeyPart().decompose(
+				keyValue,
+				0,
+				jdbcValueBindings,
+				null,
+				RowMutationOperations.DEFAULT_VALUE_SETTER,
+				session
+		);
 
 		final CollectionPart indexDescriptor = attributeMapping.getIndexDescriptor();
 
 		if ( indexDescriptor != null ) {
 			indexDescriptor.decompose(
 					incrementIndexByBase( collection.getIndex( rowValue, rowPosition, this ) ),
-					(value, jdbcValueMapping) -> {
+					0,
+					jdbcValueBindings,
+					null,
+					(valueIndex, bindings, noop, value, jdbcValueMapping) -> {
 						if ( !jdbcValueMapping.isUpdateable() ) {
 							return;
 						}
-						bindingsConsumer.consume( value, jdbcValueMapping );
+						bindings.bindValue( value, jdbcValueMapping, ParameterUsage.SET );
 					},
 					session
 			);
@@ -683,11 +698,10 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 		final EntityIdentifierMapping identifierMapping = elementDescriptor.getAssociatedEntityMappingType().getIdentifierMapping();
 		identifierMapping.decompose(
 				identifierMapping.getIdentifier( elementValue ),
-				(jdbcValue, jdbcValueMapping) -> bindingsConsumer.consumeJdbcValueBinding(
+				0,
-						jdbcValue,
+				jdbcValueBindings,
-						jdbcValueMapping,
+				null,
-						ParameterUsage.RESTRICT
+				RowMutationOperations.DEFAULT_RESTRICTOR,
-				),
 				session
 		);
 	}
@@ -754,16 +768,19 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			Object entry,
 			int entryPosition,
 			SharedSessionContractImplementor session,
-			RowMutationOperations.ValuesBindingConsumer bindingsConsumer) {
+			JdbcValueBindings jdbcValueBindings) {
 		final Object index = collection.getIndex( entry, entryPosition, this );
 
 		getAttributeMapping().getIndexDescriptor().decompose(
 				index,
-				(jdbcValue, jdbcValueMapping) -> {
+				0,
+				jdbcValueBindings,
+				null,
+				(valueIndex, bindings, noop, jdbcValue, jdbcValueMapping) -> {
 					if ( !jdbcValueMapping.isUpdateable() ) {
 						return;
 					}
-					bindingsConsumer.consume( jdbcValue, jdbcValueMapping );
+					bindings.bindValue( jdbcValue, jdbcValueMapping, ParameterUsage.SET );
 				},
 				session
 		);
@@ -775,16 +792,30 @@ public class OneToManyPersister extends AbstractCollectionPersister {
 			Object entry,
 			int entryPosition,
 			SharedSessionContractImplementor session,
-			JdbcValueConsumer jdbcValueConsumer) {
+			JdbcValueBindings jdbcValueBindings) {
 		final OneToManyCollectionPart elementDescriptor = (OneToManyCollectionPart) getAttributeMapping().getElementDescriptor();
 		final EntityMappingType associatedType = elementDescriptor.getAssociatedEntityMappingType();
 		final Object element = collection.getElement( entry );
 		final Object elementIdentifier = associatedType.getIdentifierMapping().getIdentifier( element );
-		associatedType.getIdentifierMapping().decompose( elementIdentifier, jdbcValueConsumer, session );
+		associatedType.getIdentifierMapping().decompose(
+				elementIdentifier,
+				0,
+				jdbcValueBindings,
+				null,
+				RowMutationOperations.DEFAULT_RESTRICTOR,
+				session
+		);
 
 		if ( getAttributeMapping().getIdentifierDescriptor() != null ) {
 			final Object identifier = collection.getIdentifier( entry, entryPosition );
-			getAttributeMapping().getIdentifierDescriptor().decompose( identifier, jdbcValueConsumer, session );
+			getAttributeMapping().getIdentifierDescriptor().decompose(
+					identifier,
+					0,
+					jdbcValueBindings,
+					null,
+					RowMutationOperations.DEFAULT_RESTRICTOR,
+					session
+			);
 		}
 	}
 

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

@@ -100,16 +100,7 @@ public class DeleteRowsCoordinatorStandard implements DeleteRowsCoordinator {
 						removal,
 						deletionCount,
 						session,
-						(jdbcValue, jdbcValueMapping) -> {
+						jdbcValueBindings
-							if ( jdbcValueMapping.isFormula() ) {
-								return;
-							}
-							jdbcValueBindings.bindValue(
-									jdbcValue,
-									jdbcValueMapping,
-									ParameterUsage.RESTRICT
-							);
-						}
 				);
 
 				mutationExecutor.execute( removal, null, null, null, session );

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

@@ -12,6 +12,7 @@ import org.hibernate.collection.spi.PersistentCollection;
 import org.hibernate.engine.jdbc.batch.internal.BasicBatchKey;
 import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
 import org.hibernate.engine.jdbc.mutation.MutationExecutor;
+import org.hibernate.engine.jdbc.mutation.ParameterUsage;
 import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
@@ -112,7 +113,7 @@ public class InsertRowsCoordinatorStandard implements InsertRowsCoordinator {
 							entry,
 							entryCount,
 							session,
-							jdbcValueBindings::bindValue
+							jdbcValueBindings
 					);
 					mutationExecutor.execute( entry, null, null, null, session );
 				}

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

@@ -9,7 +9,6 @@ package org.hibernate.persister.collection.mutation;
 import org.hibernate.engine.jdbc.batch.internal.BasicBatchKey;
 import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
 import org.hibernate.engine.jdbc.mutation.MutationExecutor;
-import org.hibernate.engine.jdbc.mutation.ParameterUsage;
 import org.hibernate.engine.jdbc.mutation.spi.MutationExecutorService;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
@@ -100,13 +99,10 @@ public class RemoveCoordinatorStandard implements RemoveCoordinator {
 			final ForeignKeyDescriptor fkDescriptor = mutationTarget.getTargetPart().getKeyDescriptor();
 			fkDescriptor.getKeyPart().decompose(
 					key,
-					(jdbcValue, jdbcValueMapping) -> {
+					0,
-						jdbcValueBindings.bindValue(
+					jdbcValueBindings,
-								jdbcValue,
+					null,
-								jdbcValueMapping,
+					RowMutationOperations.DEFAULT_RESTRICTOR,
-								ParameterUsage.RESTRICT
-						);
-					},
 					session
 			);
 

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

@@ -7,11 +7,11 @@
 package org.hibernate.persister.collection.mutation;
 
 import org.hibernate.collection.spi.PersistentCollection;
+import org.hibernate.engine.jdbc.mutation.JdbcValueBindings;
 import org.hibernate.engine.jdbc.mutation.ParameterUsage;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.NullnessHelper;
-import org.hibernate.metamodel.mapping.ModelPart.JdbcValueConsumer;
+import org.hibernate.metamodel.mapping.ModelPart;
-import org.hibernate.metamodel.mapping.SelectableMapping;
 import org.hibernate.sql.model.MutationOperation;
 import org.hibernate.sql.model.ast.MutatingTableReference;
 import org.hibernate.sql.model.jdbc.JdbcMutationOperation;
@@ -25,6 +25,12 @@ import org.hibernate.sql.model.jdbc.JdbcMutationOperation;
  * @author Steve Ebersole
  */
 public class RowMutationOperations {
+	public static final ModelPart.JdbcValueBiConsumer<JdbcValueBindings, Object> DEFAULT_RESTRICTOR = (valueIndex, jdbcValueBindings, o, value, jdbcValueMapping) -> {
+		jdbcValueBindings.bindValue( value, jdbcValueMapping, ParameterUsage.RESTRICT );
+	};
+	public static final ModelPart.JdbcValueBiConsumer<JdbcValueBindings, Object> DEFAULT_VALUE_SETTER = (valueIndex, jdbcValueBindings, o, value, jdbcValueMapping) -> {
+		jdbcValueBindings.bindValue( value, jdbcValueMapping, ParameterUsage.SET );
+	};
 	private final CollectionMutationTarget target;
 
 	private final OperationProducer insertRowOperationProducer;
@@ -163,7 +169,7 @@ public class RowMutationOperations {
 				Object rowValue,
 				int rowPosition,
 				SharedSessionContractImplementor session,
-				JdbcValueConsumer restrictor);
+				JdbcValueBindings jdbcValueBindings);
 	}
 
 	@FunctionalInterface
@@ -174,23 +180,7 @@ public class RowMutationOperations {
 				Object rowValue,
 				int rowPosition,
 				SharedSessionContractImplementor session,
-				ValuesBindingConsumer valuesBindingConsumer);
+				JdbcValueBindings jdbcValueBindings);
 	}
 
-	/**
-	 * Consumer for insert-value bindings
-	 *
-	 * Unfortunate we need `usage` here, but it is needed to account for
-	 * update-as-insert handling of one-to-many handling
-	 */
-	@FunctionalInterface
-	public interface ValuesBindingConsumer extends JdbcValueConsumer {
-		void consumeJdbcValueBinding(Object value, SelectableMapping jdbcValueMapping, ParameterUsage usage);
-
-		@Override
-		default void consume(Object value, SelectableMapping jdbcValueMapping) {
-			// insert values will be SET most of the time
-			consumeJdbcValueBinding( value, jdbcValueMapping, ParameterUsage.SET );
-		}
-	}
 }

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

@@ -88,11 +88,7 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 						entry,
 						entryPosition,
 						session,
-						(jdbcValue, jdbcValueMapping) -> jdbcValueBindings.bindValue(
+						jdbcValueBindings
-								jdbcValue,
-								jdbcValueMapping,
-								ParameterUsage.RESTRICT
-						)
 				);
 
 				mutationExecutor.execute( entry, null, null, null, session );
@@ -150,11 +146,7 @@ public class UpdateRowsCoordinatorOneToMany extends AbstractUpdateRowsCoordinato
 						entry,
 						entryPosition,
 						session,
-						(jdbcValue, jdbcValueMapping, usage) -> jdbcValueBindings.bindValue(
+						jdbcValueBindings
-								jdbcValue,
-								jdbcValueMapping,
-								usage
-						)
 				);
 
 				mutationExecutor.execute( entry, null, null, null, session );

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

@@ -126,11 +126,7 @@ public class UpdateRowsCoordinatorStandard extends AbstractUpdateRowsCoordinator
 				entry,
 				entryPosition,
 				session,
-				(jdbcValue, jdbcValueMapping, usage) -> mutationExecutor.getJdbcValueBindings().bindValue(
+				mutationExecutor.getJdbcValueBindings()
-						jdbcValue,
-						jdbcValueMapping,
-						usage
-				)
 		);
 
 		rowMutationOperations.getUpdateRowRestrictions().applyRestrictions(
@@ -139,11 +135,7 @@ public class UpdateRowsCoordinatorStandard extends AbstractUpdateRowsCoordinator
 				entry,
 				entryPosition,
 				session,
-				(jdbcValue, jdbcValueMapping) -> mutationExecutor.getJdbcValueBindings().bindValue(
+				mutationExecutor.getJdbcValueBindings()
-						jdbcValue,
-						jdbcValueMapping,
-						ParameterUsage.RESTRICT
-				)
 		);
 
 		mutationExecutor.execute( collection, null, null, null, session );

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

@@ -490,21 +490,36 @@ public interface EntityPersister extends EntityMappingType, RootTableGroupProduc
 	}
 
 	@Override
-	default void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	default <X, Y> int breakDownJdbcValues(
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		int span = 0;
 		if ( domainValue instanceof Object[] ) {
 			final Object[] values = (Object[]) domainValue;
 			for ( int i = 0; i < getNumberOfAttributeMappings(); i++ ) {
 				final AttributeMapping attributeMapping = getAttributeMapping( i );
-				attributeMapping.breakDownJdbcValues( values[ i ], valueConsumer, session );
+				span += attributeMapping.breakDownJdbcValues( values[ i ], offset + span, x, y, valueConsumer, session );
 			}
 		}
 		else {
 			for ( int i = 0; i < getNumberOfAttributeMappings(); i++ ) {
 				final AttributeMapping attributeMapping = getAttributeMapping( i );
 				final Object attributeValue = attributeMapping.getPropertyAccess().getGetter().get( domainValue );
-				attributeMapping.breakDownJdbcValues( attributeValue, valueConsumer, session );
+				span += attributeMapping.breakDownJdbcValues(
+						attributeValue,
+						offset + span,
+						x,
+						y,
+						valueConsumer,
+						session
+				);
 			}
 		}
+		return span;
 	}
 
 	/**

hibernate-core/src/main/java/org/hibernate/persister/entity/mutation/AbstractMutationCoordinator.java

@@ -16,6 +16,7 @@ import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.metamodel.mapping.AttributeMapping;
 import org.hibernate.persister.entity.AbstractEntityPersister;
+import org.hibernate.persister.entity.AttributeMappingsList;
 import org.hibernate.sql.model.ModelMutationLogging;
 import org.hibernate.sql.model.MutationOperation;
 import org.hibernate.sql.model.MutationOperationGroup;
@@ -112,26 +113,31 @@ public abstract class AbstractMutationCoordinator {
 			Object[] loadedState,
 			SharedSessionContractImplementor session,
 			JdbcValueBindings jdbcValueBindings) {
-		if ( entityPersister().hasPartitionedSelectionMapping() ) {
+		final AbstractEntityPersister persister = entityPersister();
-			entityPersister().forEachAttributeMapping(
+		if ( persister.hasPartitionedSelectionMapping() ) {
-					(index, attributeMapping) -> {
+			final AttributeMappingsList attributeMappings = persister.getAttributeMappings();
-						if ( attributeMapping.hasPartitionedSelectionMapping() ) {
+			final int size = attributeMappings.size();
-							attributeMapping.decompose(
+			for ( int i = 0; i < size; i++ ) {
-									loadedState[index],
+				final AttributeMapping attributeMapping = attributeMappings.get( i );
-									(value, jdbcValueMapping) -> {
+				if ( attributeMapping.hasPartitionedSelectionMapping() ) {
-										if ( jdbcValueMapping.isPartitioned() ) {
+					attributeMapping.decompose(
-											jdbcValueBindings.bindValue(
+							loadedState[i],
-													value,
+							0,
-													jdbcValueMapping,
+							jdbcValueBindings,
-													ParameterUsage.RESTRICT
+							null,
-											);
+							(valueIndex, bindings, noop, value, jdbcValueMapping) -> {
-										}
+								if ( jdbcValueMapping.isPartitioned() ) {
-									},
+									bindings.bindValue(
-									session
+											value,
-							);
+											jdbcValueMapping,
-						}
+											ParameterUsage.RESTRICT
-					}
+									);
-			);
+								}
+							},
+							session
+					);
+				}
+			}
 		}
 	}
 }

hibernate-core/src/main/java/org/hibernate/persister/entity/mutation/DeleteCoordinator.java

@@ -170,14 +170,17 @@ public class DeleteCoordinator extends AbstractMutationCoordinator {
 						final String mutationTableName = persister.getAttributeMutationTableName( attributeIndex );
 						attribute.breakDownJdbcValues(
 								loadedValue,
-								(jdbcValue, jdbcValueMapping) -> {
+								0,
+								jdbcValueBindings,
+								mutationTableName,
+								(valueIndex, bindings, tableName, jdbcValue, jdbcValueMapping) -> {
 									if ( jdbcValue == null ) {
 										// presumably the SQL was generated with `is null`
 										return;
 									}
-									jdbcValueBindings.bindValue(
+									bindings.bindValue(
 											jdbcValue,
-											mutationTableName,
+											tableName,
 											jdbcValueMapping.getSelectionExpression(),
 											ParameterUsage.RESTRICT
 									);

hibernate-core/src/main/java/org/hibernate/persister/entity/mutation/InsertCoordinator.java

@@ -243,9 +243,12 @@ public class InsertCoordinator extends AbstractMutationCoordinator {
 		if ( !(mapping instanceof PluralAttributeMapping) ) {
 			mapping.decompose(
 					value,
-					(jdbcValue, selectableMapping) -> {
+					0,
+					jdbcValueBindings,
+					null,
+					(valueIndex, bindings, noop, jdbcValue, selectableMapping) -> {
 						if ( selectableMapping.isInsertable() ) {
-							jdbcValueBindings.bindValue(
+							bindings.bindValue(
 									jdbcValue,
 									entityPersister().physicalTableNameForMutation( selectableMapping ),
 									selectableMapping.getSelectionExpression(),

hibernate-core/src/main/java/org/hibernate/persister/entity/mutation/UpdateCoordinatorStandard.java

@@ -675,10 +675,13 @@ public class UpdateCoordinatorStandard extends AbstractMutationCoordinator imple
 			Object attributeLockValue) {
 		attributeMapping.decompose(
 				attributeLockValue,
-				(jdbcValue, columnMapping) -> {
+				0,
+				analysis,
+				null,
+				(valueIndex, updateAnalysis, noop, jdbcValue, columnMapping) -> {
 					if ( !columnMapping.isFormula() ) {
 						final EntityTableMapping tableMapping = entityPersister().getPhysicalTableMappingForMutation( columnMapping );
-						analysis.registerColumnOptLock( tableMapping, columnMapping.getSelectionExpression(), jdbcValue );
+						updateAnalysis.registerColumnOptLock( tableMapping, columnMapping.getSelectionExpression(), jdbcValue );
 					}
 				},
 				session
@@ -854,11 +857,14 @@ public class UpdateCoordinatorStandard extends AbstractMutationCoordinator imple
 			Object values) {
 		attributeMapping.decompose(
 				values,
-				(jdbcValue, jdbcMapping) -> {
+				0,
+				jdbcValueBindings,
+				tableMapping,
+				(valueIndex, bindings, table, jdbcValue, jdbcMapping) -> {
 					if ( !jdbcMapping.isFormula() && jdbcMapping.isUpdateable() ) {
-						jdbcValueBindings.bindValue(
+						bindings.bindValue(
 								jdbcValue,
-								tableMapping.getTableName(),
+								table.getTableName(),
 								jdbcMapping.getSelectionExpression(),
 								ParameterUsage.SET
 						);

hibernate-core/src/main/java/org/hibernate/query/derived/AnonymousTupleBasicValuedModelPart.java

@@ -298,8 +298,15 @@ public class AnonymousTupleBasicValuedModelPart implements ModelPart, MappingTyp
 	}
 
 	@Override
-	public void breakDownJdbcValues(Object domainValue, JdbcValueConsumer valueConsumer, SharedSessionContractImplementor session) {
+	public <X, Y> int breakDownJdbcValues(
-		valueConsumer.consume( domainValue, this );
+			Object domainValue,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
+			SharedSessionContractImplementor session) {
+		valueConsumer.consume( offset, x, y, domainValue, this );
+		return getJdbcTypeCount();
 	}
 
 	@Override

hibernate-core/src/main/java/org/hibernate/query/derived/AnonymousTupleEmbeddableValuedModelPart.java

@@ -401,19 +401,23 @@ public class AnonymousTupleEmbeddableValuedModelPart implements EmbeddableValued
 	}
 
 	@Override
-	public void breakDownJdbcValues(
+	public <X, Y> int breakDownJdbcValues(
 			Object domainValue,
-			JdbcValueConsumer valueConsumer,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
 			SharedSessionContractImplementor session) {
 		final Object[] values = (Object[]) domainValue;
 		assert values.length == modelParts.size();
-
+		int span = 0;
 		int i = 0;
 		for ( ModelPart mapping : modelParts.values() ) {
 			final Object attributeValue = values[ i ];
-			mapping.breakDownJdbcValues( attributeValue, valueConsumer, session );
+			span += mapping.breakDownJdbcValues( attributeValue, offset + span, x, y, valueConsumer, session );
 			i++;
 		}
+		return span;
 	}
 
 	@Override

hibernate-core/src/main/java/org/hibernate/query/derived/AnonymousTupleEntityValuedModelPart.java

@@ -550,11 +550,14 @@ public class AnonymousTupleEntityValuedModelPart
 	}
 
 	@Override
-	public void breakDownJdbcValues(
+	public <X, Y> int breakDownJdbcValues(
 			Object domainValue,
-			JdbcValueConsumer valueConsumer,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
 			SharedSessionContractImplementor session) {
-		delegate.breakDownJdbcValues( domainValue, valueConsumer, session );
+		return delegate.breakDownJdbcValues( domainValue, offset, x, y, valueConsumer, session );
 	}
 
 	@Override

hibernate-core/src/main/java/org/hibernate/query/derived/AnonymousTupleTableGroupProducer.java

@@ -374,9 +374,12 @@ public class AnonymousTupleTableGroupProducer implements TableGroupProducer, Map
 	}
 
 	@Override
-	public void breakDownJdbcValues(
+	public <X, Y> int breakDownJdbcValues(
 			Object domainValue,
-			JdbcValueConsumer valueConsumer,
+			int offset,
+			X x,
+			Y y,
+			JdbcValueBiConsumer<X, Y> valueConsumer,
 			SharedSessionContractImplementor session) {
 		throw new UnsupportedOperationException( "Not yet implemented" );
 	}

hibernate-core/src/main/java/org/hibernate/sql/model/ast/ColumnValueBindingList.java

@@ -40,7 +40,7 @@ public class ColumnValueBindingList extends ArrayList<ColumnValueBinding> implem
 	}
 
 	@Override
-	public void consume(Object value, SelectableMapping jdbcValueMapping) {
+	public void consume(int valueIndex, Object value, SelectableMapping jdbcValueMapping) {
 		final ColumnValueBinding columnValueBinding = createValueBinding(
 				jdbcValueMapping.getSelectionExpression(),
 				value == null ? null : jdbcValueMapping.getWriteExpression(),

hibernate-envers/src/main/java/org/hibernate/envers/strategy/internal/ValidityAuditStrategy.java

@@ -33,7 +33,6 @@ import org.hibernate.envers.internal.entities.mapper.relation.MiddleComponentDat
 import org.hibernate.envers.internal.entities.mapper.relation.MiddleIdData;
 import org.hibernate.envers.internal.revisioninfo.RevisionInfoNumberReader;
 import org.hibernate.envers.internal.synchronization.SessionCacheCleaner;
-import org.hibernate.envers.internal.tools.MutableInteger;
 import org.hibernate.envers.internal.tools.query.Parameters;
 import org.hibernate.envers.internal.tools.query.QueryBuilder;
 import org.hibernate.envers.strategy.AuditStrategy;
@@ -714,34 +713,45 @@ public class ValidityAuditStrategy implements AuditStrategy {
 				int index,
 				PreparedStatement statement,
 				SessionImplementor session) {
-			final MutableInteger position = new MutableInteger( index );
+			try {
-			modelPart.breakDownJdbcValues(
+				return modelPart.breakDownJdbcValues(
-					value,
+						value,
-					(jdbcValue, jdbcValueMapping) -> {
+						index,
-						try {
+						statement,
-							//noinspection unchecked
+						session,
-							jdbcValueMapping.getJdbcMapping().getJdbcValueBinder().bind(
+						(valueIndex, preparedStatement, sessionImplementor, jdbcValue, jdbcValueMapping) -> {
-									statement,
+							try {
-									jdbcValue,
+								//noinspection unchecked
-									position.getAndIncrease(),
+								jdbcValueMapping.getJdbcMapping().getJdbcValueBinder().bind(
-									session
+										preparedStatement,
-							);
+										jdbcValue,
-						}
+										valueIndex,
-						catch (SQLException e) {
+										sessionImplementor
-							throw session.getJdbcServices().getSqlExceptionHelper().convert(
+								);
-									e,
+							}
-									String.format(
+							catch (SQLException e) {
-											Locale.ROOT,
+								throw new NestedRuntimeException( e );
-											"Error binding JDBC value relative to `%s`",
+							}
-											modelPart.getNavigableRole().getFullPath()
+						},
-									)
+						session
-							);
+				);
-						}
+			}
-					},
+			catch (NestedRuntimeException e) {
-					session
+				throw session.getJdbcServices().getSqlExceptionHelper().convert(
-			);
+						(SQLException) e.getCause(),
+						String.format(
+								Locale.ROOT,
+								"Error binding JDBC value relative to `%s`",
+								modelPart.getNavigableRole().getFullPath()
+						)
+				);
+			}
+		}
 
-			return modelPart.getJdbcTypeCount();
+		static class NestedRuntimeException extends RuntimeException {
+			public NestedRuntimeException(SQLException cause) {
+				super( cause );
+			}
 		}
 	}
 }