HHH-16959 Fail to batch delete entities with nested embeddeds

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

@@ -65,6 +65,7 @@ public abstract class AbstractMutationCoordinator {
 
 	protected BatchKeyAccess resolveBatchKeyAccess(boolean dynamicUpdate, SharedSessionContractImplementor session) {
 		if ( !dynamicUpdate
+				&& !entityPersister().optimisticLockStyle().isAllOrDirty()
 				&& session.getTransactionCoordinator() != null
 				&& session.getTransactionCoordinator().isTransactionActive() ) {
 			return this::getBatchKey;

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

@@ -26,6 +26,8 @@ import org.hibernate.persister.entity.AbstractEntityPersister;
 import org.hibernate.sql.model.MutationOperation;
 import org.hibernate.sql.model.MutationOperationGroup;
 import org.hibernate.sql.model.MutationType;
+import org.hibernate.sql.model.ast.ColumnValueBinding;
+import org.hibernate.sql.model.ast.ColumnValueBindingList;
 import org.hibernate.sql.model.ast.builder.MutationGroupBuilder;
 import org.hibernate.sql.model.ast.builder.RestrictedTableMutationBuilder;
 import org.hibernate.sql.model.ast.builder.TableDeleteBuilder;
@@ -439,18 +441,26 @@ public class DeleteCoordinator extends AbstractMutationCoordinator {
 			Object loadedValue) {
 		final RestrictedTableMutationBuilder<?, ?> tableMutationBuilder =
 				mutationGroupBuilder.findTableDetailsBuilder( attribute.getContainingTableExpression() );
-		if ( tableMutationBuilder != null && tableMutationBuilder.getOptimisticLockBindings() != null ) {
+		if ( tableMutationBuilder != null ) {
+			final ColumnValueBindingList optimisticLockBindings = tableMutationBuilder.getOptimisticLockBindings();
+			if ( optimisticLockBindings != null ) {
 				attribute.breakDownJdbcValues(
 						loadedValue,
 						(valueIndex, value, jdbcValueMapping) -> {
-						if ( value != null && !tableMutationBuilder.getKeyRestrictionBindings().contains( value ) ) {
+							final ColumnValueBinding valueBinding = optimisticLockBindings.createValueBinding(
-							tableMutationBuilder.getOptimisticLockBindings().consume( valueIndex, value, jdbcValueMapping );
+									jdbcValueMapping.getSelectableName(),
+									value == null ? null : jdbcValueMapping.getWriteExpression(),
+									jdbcValueMapping.getJdbcMapping()
+							);
+							if ( !tableMutationBuilder.getKeyRestrictionBindings().contains( valueBinding ) ) {
+								optimisticLockBindings.add( valueBinding );
 							}
 						}
 						,
 						session
 				);
 			}
+		}
 		// else there is no actual delete statement for that table,
 		// generally indicates we have an on-delete=cascade situation
 	}

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

@@ -6,6 +6,8 @@
  */
 package org.hibernate.sql.model.ast;
 
+import java.util.Objects;
+
 import org.hibernate.sql.ast.tree.expression.ColumnReference;
 
 /**
@@ -38,4 +40,21 @@ public class ColumnValueBinding {
 	public String toString() {
 		return "ColumnValueBinding(" + valueExpression + ")";
 	}
+
+	@Override
+	public boolean equals(Object o) {
+		if ( this == o ) {
+			return true;
+		}
+		if ( o == null || getClass() != o.getClass() ) {
+			return false;
+		}
+		ColumnValueBinding that = (ColumnValueBinding) o;
+		return Objects.equals( columnReference, that.columnReference );
+	}
+
+	@Override
+	public int hashCode() {
+		return Objects.hash( columnReference );
+	}
 }

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

@@ -54,7 +54,7 @@ public class ColumnValueBindingList extends ArrayList<ColumnValueBinding> implem
 		add( createValueBinding( columnName, columnWriteFragment, jdbcMapping ) );
 	}
 
-	protected ColumnValueBinding createValueBinding(
+	public ColumnValueBinding createValueBinding(
 			String columnName,
 			String customWriteExpression,
 			JdbcMapping jdbcMapping) {