HHH-16959 Fail to batch delete entities with nested embeddeds

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

@@ -70,6 +70,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

@@ -25,6 +25,8 @@ import org.hibernate.metamodel.mapping.SelectableMapping;
 import org.hibernate.persister.entity.AbstractEntityPersister;
 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;
@@ -434,18 +436,25 @@ 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 ) ) {
+							if ( !tableMutationBuilder.getKeyRestrictionBindings()
-							tableMutationBuilder.getOptimisticLockBindings().consume( valueIndex, value, jdbcValueMapping );
+									.containsColumn(
+											jdbcValueMapping.getSelectableName(),
+											jdbcValueMapping.getJdbcMapping()
+									) ) {
+								optimisticLockBindings.consume( valueIndex, value, jdbcValueMapping );
 							}
 						}
 						,
 						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

@@ -13,6 +13,7 @@ import org.hibernate.engine.jdbc.mutation.ParameterUsage;
 import org.hibernate.metamodel.mapping.JdbcMapping;
 import org.hibernate.metamodel.mapping.ModelPart;
 import org.hibernate.metamodel.mapping.SelectableMapping;
+import org.hibernate.sql.ast.tree.expression.ColumnReference;
 import org.hibernate.sql.model.ast.builder.ColumnValueBindingBuilder;
 
 @Internal
@@ -68,6 +69,16 @@ public class ColumnValueBindingList extends ArrayList<ColumnValueBinding> implem
 		);
 	}
 
+	public boolean containsColumn(String columnName, JdbcMapping jdbcMapping) {
+		final ColumnReference reference = new ColumnReference( mutatingTable, columnName, jdbcMapping );
+		for ( int i = 0; i < size(); i++ ) {
+			if ( get( i ).getColumnReference().equals( reference ) ) {
+				return true;
+			}
+		}
+		return false;
+	}
+
 	@Override
 	public String toString() {
 		return "ColumnValueBindingList" + super.toString();