HHH-16281 Inconsistent Behaivor of L2 cache between Hibernate 5 and 6

hibernate-core/src/main/java/org/hibernate/cache/MutableCacheKeyBuilder.java

@@ -0,0 +1,31 @@
+/*
+ * 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.cache;
+
+import java.io.Serializable;
+
+import org.hibernate.cache.spi.QueryResultsCache;
+
+/**
+ * A builder that generates a Serializable Object to be used as a key into the {@linkplain QueryResultsCache
+ * query results cache}.
+ */
+
+public interface MutableCacheKeyBuilder extends Serializable {
+
+	void addValue(Object value);
+
+
+	void addHashCode(int hashCode);
+
+	/**
+	 *  creates an Object to be used as a key into the {@linkplain QueryResultsCache
+	 *  query results cache}.
+	 */
+	Serializable build();
+
+}

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

@@ -6,9 +6,15 @@
  */
 package org.hibernate.metamodel.mapping;
 
+import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.type.descriptor.converter.spi.BasicValueConverter;
+import org.hibernate.type.descriptor.java.JavaType;
+
 /**
  * Any basic-typed ValueMapping.  Generally this would be one of<ul>
  *     <li>a {@link jakarta.persistence.Basic} attribute</li>
@@ -43,4 +49,29 @@ public interface BasicValuedMapping extends ValueMapping, SqlExpressible {
 	}
 
 	JdbcMapping getJdbcMapping();
+
+	@Override
+	default Object disassemble(Object value, SharedSessionContractImplementor session) {
+		return getJdbcMapping().convertToRelationalValue( value );
+	}
+
+	@Override
+	default void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session){
+		final JdbcMapping jdbcMapping = getJdbcMapping();
+		final BasicValueConverter converter = jdbcMapping.getValueConverter();
+		final Serializable disassemble;
+		final int hashCode;
+		if ( converter == null ) {
+			disassemble = jdbcMapping.getJavaTypeDescriptor().getMutabilityPlan().disassemble( value, session );
+			hashCode = ( (JavaType) jdbcMapping.getMappedJavaType() ).extractHashCode( value );
+		}
+		else {
+			final Object relationalValue = converter.toRelationalValue( value );
+			final JavaType relationalJavaType = converter.getRelationalJavaType();
+			disassemble = relationalJavaType.getMutabilityPlan().disassemble( relationalValue, session );
+			hashCode = relationalJavaType.extractHashCode( relationalValue );
+		}
+		cacheKey.addValue( disassemble );
+		cacheKey.addHashCode( hashCode );
+	}
 }

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

@@ -10,6 +10,7 @@ import java.util.ArrayList;
 import java.util.List;
 
 import org.hibernate.Incubating;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 
@@ -51,52 +52,56 @@ public interface Bindable extends JdbcMappingContainer {
 	}
 
 	/**
-	 * @asciidoc
+	 * @asciidoc Breaks down a value of `J` into its simple pieces.  E.g., an embedded
-	 *
-	 * Breaks down a value of `J` into its simple pieces.  E.g., an embedded
 	 * value gets broken down into an array of its attribute state; a basic
 	 * value converts to itself; etc.
 	 * <p>
 	 * Generally speaking, this is the form in which entity state is kept relative to a
 	 * Session via `EntityEntry`.
-	 *
+	 * @Entity class Person {
-	 * @see org.hibernate.engine.spi.EntityEntry
+	 * @Id Integer id;
-	 *
+	 * @Embedded Name name;
-	 * As an example, consider the following domain model:
+	 * int age;
-	 *
-	 * ````
-	 * @Entity
-	 * class Person {
-	 * 		@Id Integer id;
-	 * 		@Embedded Name name;
-	 * 		int age;
 	 * }
-	 *
+	 * @Embeddable class Name {
-	 * @Embeddable
+	 * String familiarName;
-	 * class Name {
+	 * String familyName;
-	 *     String familiarName;
-	 *     String familyName;
 	 * }
 	 * ````
-	 *
+	 * <p>
 	 * At the top-level, we would want to disassemble a `Person` value so we'd ask the
 	 * `Bindable` for the `Person` entity to disassemble.  Given a Person value:
-	 *
+	 * <p>
 	 * ````
 	 * Person( id=1, name=Name( 'Steve', 'Ebersole' ), 28 )
 	 * ````
-	 *
+	 * <p>
 	 * this disassemble would result in a multi-dimensional array:
-	 *
+	 * <p>
 	 * ````
 	 * [ ["Steve", "Ebersole"], 28 ]
 	 * ````
-	 *
+	 * <p>
 	 * Note that the identifier is not part of this disassembled state.  Note also
 	 * how the embedded value results in a sub-array.
+	 * @see org.hibernate.engine.spi.EntityEntry
+	 * <p>
+	 * As an example, consider the following domain model:
+	 * <p>
+	 * ````
 	 */
 	Object disassemble(Object value, SharedSessionContractImplementor session);
 
+	/**
+	 * Add to the MutableCacheKey the values obtained disassembling the value and the hasCode generated from
+	 * the disassembled value.
+	 *
+	 * @param cacheKey the MutableCacheKey used to add the disassembled value and the hashCode
+	 * @param value the value to disassemble
+	 * @param session the SharedSessionContractImplementor
+	 */
+	void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session);
+
 	/**
 	 * @asciidoc
 	 *

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping;
 import java.util.List;
 import java.util.function.Consumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.property.access.spi.PropertyAccess;
@@ -115,6 +116,11 @@ public interface EmbeddableValuedModelPart extends ValuedModelPart, Fetchable, F
 		return getEmbeddableTypeDescriptor().disassemble( value, session );
 	}
 
+	@Override
+	default void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		getEmbeddableTypeDescriptor().addToCacheKey( cacheKey, value, session );
+	}
+
 	/**
 	 * @see org.hibernate.annotations.Parent
 	 */

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.spi.NavigablePath;
@@ -100,6 +101,11 @@ public interface EntityValuedModelPart extends FetchableContainer {
 		return getEntityMappingType().disassemble( value, session );
 	}
 
+	@Override
+	default void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session){
+		getEntityMappingType().addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	default <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -14,6 +14,7 @@ import java.util.function.Consumer;
 import org.hibernate.MappingException;
 import org.hibernate.SharedSessionContract;
 import org.hibernate.bytecode.spi.ReflectionOptimizer;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.dialect.Dialect;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.jdbc.env.spi.JdbcEnvironment;
@@ -575,11 +576,10 @@ public abstract class AbstractEmbeddableMapping implements EmbeddableMappingType
 
 	@Override
 	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		final MutableAttributeMappingList attributes = attributeMappings;
+		final int size = attributeMappings.size();
-		final int size = attributes.size();
 		final Object[] result = new Object[ size ];
 		for ( int i = 0; i < size; i++ ) {
-			final AttributeMapping attributeMapping = attributes.get( i );
+			final AttributeMapping attributeMapping = attributeMappings.get( i );
 			final Object o = attributeMapping.getValue( value );
 			result[i] = attributeMapping.disassemble( o, session );
 		}
@@ -587,6 +587,15 @@ public abstract class AbstractEmbeddableMapping implements EmbeddableMappingType
 		return result;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final int size = attributeMappings.size();
+		for ( int i = 0; i < size; i++ ) {
+			final AttributeMapping attributeMapping = attributeMappings.get( i );
+			attributeMapping.addToCacheKey( cacheKey, attributeMapping.getValue( value ), session );
+		}
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -6,9 +6,9 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
-import java.io.Serializable;
 import java.util.function.BiConsumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
@@ -209,8 +209,13 @@ public class AnyDiscriminatorPart implements BasicValuedModelPart, FetchOptions,
 
 	@Override
 	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		final Serializable discriminator = metaType.disassemble( value, session, value );
+		return metaType.disassemble( value, session, value );
-		return discriminator;
+	}
+
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		cacheKey.addValue( metaType.disassemble( value, session, value ) );
+		cacheKey.addHashCode( metaType.getHashCode( value ) );
 	}
 
 	@Override

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

@@ -6,6 +6,7 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
+import java.io.Serializable;
 import java.util.function.BiConsumer;
 
 import org.hibernate.engine.FetchStyle;
@@ -393,11 +394,6 @@ public class BasicAttributeMapping
 		);
 	}
 
-	@Override
-	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		return jdbcMapping.convertToRelationalValue( value );
-	}
-
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -10,6 +10,7 @@ import java.util.Locale;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.internal.UnsavedValueFactory;
@@ -382,6 +383,11 @@ public class BasicEntityIdentifierMappingImpl implements BasicEntityIdentifierMa
 		return idType.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		idType.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -6,6 +6,7 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
+import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.BiConsumer;
@@ -357,9 +358,4 @@ public class BasicValuedCollectionPart
 		valuesConsumer.consume( offset, x, y, value, getJdbcMapping() );
 		return getJdbcTypeCount();
 	}
-
-	@Override
-	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		return selectableMapping.getJdbcMapping().convertToRelationalValue( value );
-	}
 }

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

@@ -8,6 +8,7 @@ package org.hibernate.metamodel.mapping.internal;
 
 import java.util.function.BiConsumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
@@ -323,6 +324,11 @@ public class CollectionIdentifierDescriptorImpl implements CollectionIdentifierD
 		return type.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		type.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -14,6 +14,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.hibernate.HibernateException;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -393,6 +394,18 @@ public class CompoundNaturalIdMapping extends AbstractNaturalIdMapping implement
 		return outgoing;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		assert value instanceof Object[];
+
+		final Object[] values = (Object[]) value;
+		assert values.length == attributes.size();
+
+		for ( int i = 0; i < attributes.size(); i++ ) {
+			attributes.get( i ).addToCacheKey( cacheKey, values[i], session );
+		}
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -12,6 +12,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.hibernate.SharedSessionContract;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
@@ -235,6 +236,23 @@ public class DiscriminatedAssociationAttributeMapping
 		};
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		if ( value == null ) {
+			return ;
+		}
+
+		final EntityMappingType concreteMappingType = determineConcreteType( value, session );
+
+		final Object discriminator = discriminatorMapping
+				.getModelPart()
+				.resolveDiscriminatorForEntityType( concreteMappingType );
+		discriminatorMapping.getDiscriminatorPart().addToCacheKey( cacheKey, discriminator, session );
+
+		final EntityIdentifierMapping identifierMapping = concreteMappingType.getIdentifierMapping();
+		identifierMapping.addToCacheKey( cacheKey, identifierMapping.getIdentifier( value ), session );
+	}
+
 	private EntityMappingType determineConcreteType(Object entity, SharedSessionContractImplementor session) {
 		final String entityName;
 		if ( session == null ) {

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping.internal;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
@@ -264,6 +265,11 @@ public class DiscriminatedCollectionPart implements DiscriminatedAssociationMode
 		return discriminatorMapping.getDiscriminatorPart().disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		discriminatorMapping.getDiscriminatorPart().addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -11,6 +11,7 @@ import java.util.List;
 import java.util.function.BiConsumer;
 import java.util.function.IntFunction;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.internal.util.MutableInteger;
@@ -603,6 +604,11 @@ public class EmbeddedForeignKeyDescriptor implements ForeignKeyDescriptor {
 		return targetSide.getModelPart().disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		targetSide.getModelPart().addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public boolean hasPartitionedSelectionMapping() {
 		return keySide.getModelPart().hasPartitionedSelectionMapping();

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

@@ -6,6 +6,7 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
+import java.io.Serializable;
 import java.util.function.BiConsumer;
 
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -158,17 +159,4 @@ public class EmbeddedIdentifierMappingImpl
 				session
 		);
 	}
-
-	@Override
-	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		final EmbeddableMappingType embeddableTypeDescriptor = getEmbeddableTypeDescriptor();
-		final Object[] result = new Object[embeddableTypeDescriptor.getNumberOfAttributeMappings()];
-		embeddableTypeDescriptor.forEachAttributeMapping(
-				(i, mapping) -> {
-					Object o = mapping.getPropertyAccess().getGetter().get( value );
-					result[i] = mapping.disassemble( o, session );
-				}
-		);
-		return result;
-	}
 }

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

@@ -8,6 +8,8 @@ package org.hibernate.metamodel.mapping.internal;
 
 import org.hibernate.Internal;
 import org.hibernate.annotations.NotFoundAction;
+import org.hibernate.cache.MutableCacheKeyBuilder;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.mapping.Collection;
 import org.hibernate.metamodel.mapping.CollectionPart;
 import org.hibernate.metamodel.mapping.EntityMappingType;
@@ -65,6 +67,11 @@ public interface EntityCollectionPart extends CollectionPart, EntityValuedFetcha
 		return CollectionPart.super.getJdbcTypeCount();
 	}
 
+	@Override
+	default void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		EntityValuedFetchable.super.addToCacheKey( cacheKey, value, session );
+	}
+
 	/**
 	 * Perform any delayed initialization.
 	 * <p>

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

@@ -8,6 +8,7 @@ package org.hibernate.metamodel.mapping.internal;
 
 import java.util.function.BiConsumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
@@ -123,6 +124,11 @@ public class EntityRowIdMappingImpl implements EntityRowIdMapping {
 		return rowIdType.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		rowIdType.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping.internal;
 import java.util.function.BiConsumer;
 import java.util.function.Supplier;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.internal.UnsavedValueFactory;
@@ -334,6 +335,11 @@ public class EntityVersionMappingImpl implements EntityVersionMapping, FetchOpti
 		return versionBasicType.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		versionBasicType.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -6,8 +6,10 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
+import java.io.Serializable;
 import java.util.function.Consumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.EntityKey;
@@ -303,7 +305,6 @@ public class IdClassEmbeddable extends AbstractEmbeddableMapping implements Iden
 		return span;
 	}
 
-
 	@Override
 	public Object disassemble(Object value, SharedSessionContractImplementor session) {
 		// todo (6.0) : reduce to-one values to id here?
@@ -317,6 +318,16 @@ public class IdClassEmbeddable extends AbstractEmbeddableMapping implements Iden
 		return result;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final Serializable[] result = new Serializable[ getNumberOfAttributeMappings() ];
+		for ( int i = 0; i < result.length; i++ ) {
+			final AttributeMapping attributeMapping = getAttributeMapping( i );
+			final Object o = attributeMapping.getPropertyAccess().getGetter().get( value );
+			attributeMapping.addToCacheKey( cacheKey, o, session );
+		}
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping.internal;
 import java.util.List;
 import java.util.function.BiConsumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -133,6 +134,11 @@ public class InverseNonAggregatedIdentifierMapping extends EmbeddedAttributeMapp
 		return identifierValueMapper.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		identifierValueMapper.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachJdbcValue(
 			Object value,

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

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.mapping.internal;
 import java.util.List;
 import java.util.function.BiConsumer;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -147,6 +148,11 @@ public class NonAggregatedIdentifierMappingImpl extends AbstractCompositeIdentif
 		return identifierValueMapper.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		identifierValueMapper.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachJdbcValue(
 			Object value,

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

@@ -105,16 +105,6 @@ public class OneToManyCollectionPart extends AbstractEntityCollectionPart implem
 		);
 	}
 
-	@Override
-	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		if ( value == null ) {
-			return null;
-		}
-
-		// should be an instance of the associated entity
-		return getAssociatedEntityMappingType().getIdentifierMapping().getIdentifier( value );
-	}
-
 	@Override
 	public String getContainingTableExpression() {
 		return getCollectionDescriptor().getAttributeMapping().getKeyDescriptor().getContainingTableExpression();

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

@@ -10,6 +10,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.CascadeStyle;
@@ -995,7 +996,12 @@ public class PluralAttributeMappingImpl
 
 	@Override
 	public Object disassemble(Object value, SharedSessionContractImplementor session) {
-		return elementDescriptor.disassemble( value,session );
+		return elementDescriptor.disassemble( value, session );
+	}
+
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		elementDescriptor.addToCacheKey( cacheKey, value, session );
 	}
 
 	@Override

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

@@ -6,12 +6,14 @@
  */
 package org.hibernate.metamodel.mapping.internal;
 
+import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.function.BiConsumer;
 import java.util.function.IntFunction;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -52,6 +54,7 @@ import org.hibernate.sql.results.graph.Fetch;
 import org.hibernate.sql.results.graph.FetchOptions;
 import org.hibernate.sql.results.graph.FetchParent;
 import org.hibernate.sql.results.graph.basic.BasicResult;
+import org.hibernate.type.descriptor.converter.spi.BasicValueConverter;
 import org.hibernate.type.descriptor.java.JavaType;
 
 /**
@@ -418,6 +421,28 @@ public class SimpleForeignKeyDescriptor implements ForeignKeyDescriptor, BasicVa
 		return getJdbcMapping().convertToRelationalValue( value );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final JdbcMapping jdbcMapping = getJdbcMapping();
+		final BasicValueConverter converter = jdbcMapping.getValueConverter();
+		final Serializable disassemble;
+		final int hashCode;
+		if ( converter == null ) {
+			final JavaType javaTypeDescriptor = jdbcMapping.getJavaTypeDescriptor();
+			disassemble = javaTypeDescriptor.getMutabilityPlan().disassemble( value, session );
+			hashCode = javaTypeDescriptor.extractHashCode( disassemble );
+		}
+		else {
+			final Object relationalValue = converter.toRelationalValue( value );
+			final JavaType relationalJavaType = converter.getRelationalJavaType();
+			disassemble = relationalJavaType.getMutabilityPlan().disassemble( relationalValue, session );
+			hashCode = relationalJavaType.extractHashCode( relationalValue );
+		}
+
+		cacheKey.addValue( disassemble );
+		cacheKey.addHashCode( hashCode );
+	}
+
 	@Override
 	public Object getAssociationKeyFromSide(
 			Object targetObject,

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

@@ -13,6 +13,7 @@ import java.util.Map;
 import java.util.function.BiConsumer;
 
 import org.hibernate.HibernateException;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
@@ -242,6 +243,11 @@ public class SimpleNaturalIdMapping extends AbstractNaturalIdMapping implements
 		return attribute.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		attribute.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int breakDownJdbcValues(
 			Object domainValue,

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

@@ -16,6 +16,7 @@ import java.util.function.Supplier;
 
 import org.hibernate.LockMode;
 import org.hibernate.annotations.NotFoundAction;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
@@ -2241,6 +2242,11 @@ public class ToOneAttributeMapping
 		);
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		foreignKeyDescriptor.addToCacheKey( cacheKey, foreignKeyDescriptor.getAssociationKeyFromSide( value, sideNature.inverse(), session ), session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

hibernate-core/src/main/java/org/hibernate/metamodel/model/domain/internal/ArrayTupleType.java

@@ -9,6 +9,7 @@ package org.hibernate.metamodel.model.domain.internal;
 import java.util.Arrays;
 import java.util.List;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.metamodel.UnsupportedMappingException;
@@ -93,6 +94,11 @@ public class ArrayTupleType implements TupleType<Object[]>,
 		throw new UnsupportedOperationException();
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		throw new UnsupportedOperationException();
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

hibernate-core/src/main/java/org/hibernate/metamodel/model/domain/internal/TupleMappingModelExpressible.java

@@ -6,6 +6,7 @@
  */
 package org.hibernate.metamodel.model.domain.internal;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.metamodel.mapping.JdbcMapping;
@@ -41,6 +42,13 @@ public class TupleMappingModelExpressible implements MappingModelExpressible {
 		return disassembled;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		for ( int i = 0; i < components.length; i++ ) {
+			components[i].addToCacheKey( cacheKey, value, session );
+		}
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -15,6 +15,7 @@ import java.util.Objects;
 import org.hibernate.HibernateException;
 import org.hibernate.Internal;
 import org.hibernate.MappingException;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.Mapping;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -195,6 +196,11 @@ public class DiscriminatorType<T> extends AbstractType implements BasicType<T>,
 		return toRelationalValue( (T) value );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		underlyingType.addToCacheKey( cacheKey, value, session );
+	}
+
 	// simple delegation ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 	@Override

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

@@ -17,6 +17,7 @@ import org.hibernate.LockOptions;
 import org.hibernate.MappingException;
 import org.hibernate.bytecode.enhance.spi.interceptor.EnhancementAsProxyLazinessInterceptor;
 import org.hibernate.bytecode.spi.BytecodeEnhancementMetadata;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.cache.spi.access.EntityDataAccess;
 import org.hibernate.cache.spi.access.NaturalIdDataAccess;
 import org.hibernate.cache.spi.entry.CacheEntry;
@@ -1003,6 +1004,14 @@ public interface EntityPersister extends EntityMappingType, RootTableGroupProduc
 		return this;
 	}
 
+	@Override
+	default void addToCacheKey(
+			MutableCacheKeyBuilder cacheKey,
+			Object value,
+			SharedSessionContractImplementor session) {
+		getIdentifierMapping().addToCacheKey( cacheKey, value, session );
+	}
+
 	BytecodeEnhancementMetadata getInstrumentationMetadata();
 
 	default BytecodeEnhancementMetadata getBytecodeEnhancementMetadata() {

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

@@ -9,6 +9,7 @@ package org.hibernate.query.derived;
 import java.util.function.BiConsumer;
 
 import org.hibernate.Incubating;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -314,6 +315,12 @@ public class AnonymousTupleBasicValuedModelPart implements ModelPart, MappingTyp
 		return value;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		cacheKey.addValue( value );
+		cacheKey.addHashCode( ( (JavaType) getExpressibleJavaType() ).extractHashCode( value ) );
+	}
+
 	@Override
 	public <X, Y> int forEachJdbcValue(
 			Object value,

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

@@ -13,6 +13,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.hibernate.Incubating;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.FetchStyle;
 import org.hibernate.engine.FetchTiming;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -425,6 +426,16 @@ public class AnonymousTupleEmbeddableValuedModelPart implements EmbeddableValued
 		return result;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final Object[] values = (Object[]) value;
+		int i = 0;
+		for ( ModelPart mapping : modelParts.values() ) {
+			mapping.addToCacheKey( cacheKey, values[i], session );
+			i++;
+		}
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -14,6 +14,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.hibernate.Incubating;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.OptimisticLockStyle;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
@@ -551,6 +552,11 @@ public class AnonymousTupleEntityValuedModelPart
 		return delegate.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		delegate.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

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

@@ -14,6 +14,7 @@ import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 
 import org.hibernate.Incubating;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.internal.util.collections.CollectionHelper;
@@ -389,6 +390,11 @@ public class AnonymousTupleTableGroupProducer implements TableGroupProducer, Map
 		throw new UnsupportedOperationException( "Not yet implemented" );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		throw new UnsupportedOperationException( "Not yet implemented" );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

hibernate-core/src/main/java/org/hibernate/query/internal/QueryParameterBindingsImpl.java

@@ -18,6 +18,7 @@ import java.util.function.BiConsumer;
 import org.hibernate.Incubating;
 import org.hibernate.QueryException;
 import org.hibernate.QueryParameterException;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.cache.spi.QueryKey;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -166,45 +167,25 @@ public class QueryParameterBindingsImpl implements QueryParameterBindings {
 
 	@Override
 	public QueryKey.ParameterBindingsMemento generateQueryKeyMemento(SharedSessionContractImplementor persistenceContext) {
-		final int size = parameterBindingMap.size();
+		final MutableCacheKeyImpl mutableCacheKey = new MutableCacheKeyImpl(parameterBindingMap.size());
-		final List<Object> allBindValues = new ArrayList<>( size );
-		int hashCode = 0;
 
 		for ( QueryParameterBinding<?> binding : parameterBindingMap.values() ) {
 			final MappingModelExpressible<?> mappingType = determineMappingType( binding, persistenceContext );
 			assert mappingType instanceof JavaTypedExpressible;
-			//noinspection unchecked
-			final JavaType<Object> javaType = ( (JavaTypedExpressible<Object>) mappingType ).getExpressibleJavaType();
 
 			if ( binding.isMultiValued() ) {
 				for ( Object bindValue : binding.getBindValues() ) {
 					assert bindValue != null;
-
+					mappingType.addToCacheKey( mutableCacheKey, bindValue, persistenceContext );
-					final Object disassembled = mappingType.disassemble( bindValue, persistenceContext );
-					allBindValues.add( disassembled );
-
-					final int valueHashCode = bindValue != null
-							? javaType.extractHashCode( bindValue )
-							: 0;
-
-					hashCode = 37 * hashCode + valueHashCode;
 				}
 			}
 			else {
 				final Object bindValue = binding.getBindValue();
-
+				mappingType.addToCacheKey( mutableCacheKey, bindValue, persistenceContext );
-				final Object disassembled = mappingType.disassemble( bindValue, persistenceContext );
-				allBindValues.add( disassembled );
-
-				final int valueHashCode = bindValue != null
-						? javaType.extractHashCode( bindValue )
-						: 0;
-
-				hashCode = 37 * hashCode + valueHashCode;
 			}
 		}
 
-		return new ParameterBindingsMementoImpl( allBindValues.toArray( new Object[0] ), hashCode );
+		return mutableCacheKey.build();
 	}
 
 	private MappingModelExpressible<?> determineMappingType(QueryParameterBinding<?> binding, SharedSessionContractImplementor session) {
@@ -250,10 +231,34 @@ public class QueryParameterBindingsImpl implements QueryParameterBindings {
 		return typeConfiguration.getBasicTypeForJavaType( binding.getBindType().getBindableJavaType() );
 	}
 
+	private static class MutableCacheKeyImpl implements MutableCacheKeyBuilder {
+
+		final List<Object> values;
+		int hashCode;
+
+		public MutableCacheKeyImpl(int parameterBindingMapSize) {
+			values = new ArrayList<>( parameterBindingMapSize );
+		}
+
+		@Override
+		public void addValue(Object value) {
+			values.add( value );
+		}
+
+		@Override
+		public void addHashCode(int hashCode) {
+			this.hashCode = 37 * this.hashCode + hashCode;
+		}
+
+		@Override
+		public QueryKey.ParameterBindingsMemento build() {
+			return new ParameterBindingsMementoImpl( values.toArray( new Object[0] ), hashCode );
+		}
+	}
 
 	private static class ParameterBindingsMementoImpl implements QueryKey.ParameterBindingsMemento {
-		private final Object[] values;
+		final Object[] values;
-		private final int hashCode;
+		final int hashCode;
 
 		private ParameterBindingsMementoImpl(Object[] values, int hashCode) {
 			this.values = values;
@@ -274,8 +279,7 @@ public class QueryParameterBindingsImpl implements QueryParameterBindings {
 			if ( hashCode != queryKey.hashCode ) {
 				return false;
 			}
-			// Probably incorrect - comparing Object[] arrays with Arrays.equals
+			return Arrays.deepEquals( values, queryKey.values );
-			return Arrays.equals( values, queryKey.values );
 		}
 
 		@Override

hibernate-core/src/main/java/org/hibernate/sql/ast/tree/expression/EntityTypeLiteral.java

@@ -8,6 +8,7 @@ package org.hibernate.sql.ast.tree.expression;
 
 import java.util.List;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.metamodel.mapping.JdbcMapping;
@@ -79,6 +80,11 @@ public class EntityTypeLiteral
 		return discriminatorType.disassemble( value, session );
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		discriminatorType.addToCacheKey( cacheKey, value, session );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

hibernate-core/src/main/java/org/hibernate/sql/ast/tree/expression/JdbcLiteral.java

@@ -6,11 +6,13 @@
  */
 package org.hibernate.sql.ast.tree.expression;
 
+import java.io.Serializable;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.util.Collections;
 import java.util.List;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.metamodel.mapping.JdbcMapping;
@@ -27,6 +29,7 @@ import org.hibernate.sql.results.graph.DomainResultCreationState;
 import org.hibernate.sql.results.graph.basic.BasicResult;
 import org.hibernate.type.descriptor.java.JavaType;
 import org.hibernate.type.descriptor.java.JavaTypedExpressible;
+import org.hibernate.type.descriptor.java.MutabilityPlan;
 
 /**
  * Represents a literal in the SQL AST.  This form accepts a {@link JdbcMapping} and acts
@@ -118,6 +121,15 @@ public class JdbcLiteral<T> implements Literal, MappingModelExpressible<T>, Doma
 		return value;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final Serializable disassemble = ( (MutabilityPlan<Object>) jdbcMapping.getJdbcJavaType().getMutabilityPlan() )
+				.disassemble( value, session );
+		final int hashCode = jdbcMapping.getJavaTypeDescriptor().extractHashCode( value );
+		cacheKey.addValue( disassemble );
+		cacheKey.addHashCode( hashCode );
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

hibernate-core/src/main/java/org/hibernate/sql/exec/internal/AbstractJdbcParameter.java

@@ -9,6 +9,7 @@ package org.hibernate.sql.exec.internal;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.IndexedConsumer;
 import org.hibernate.metamodel.mapping.JdbcMapping;
@@ -24,6 +25,7 @@ import org.hibernate.sql.exec.spi.JdbcParameterBinder;
 import org.hibernate.sql.exec.spi.JdbcParameterBinding;
 import org.hibernate.sql.exec.spi.JdbcParameterBindings;
 import org.hibernate.sql.results.internal.SqlSelectionImpl;
+import org.hibernate.type.descriptor.converter.spi.BasicValueConverter;
 import org.hibernate.type.descriptor.java.JavaType;
 import org.hibernate.type.spi.TypeConfiguration;
 
@@ -154,6 +156,24 @@ public abstract class AbstractJdbcParameter
 		return value;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final JdbcMapping jdbcMapping = getJdbcMapping();
+		final BasicValueConverter converter = jdbcMapping.getValueConverter();
+
+		if ( converter == null ) {
+			final JavaType javaTypeDescriptor = jdbcMapping.getJavaTypeDescriptor();
+			cacheKey.addValue( javaTypeDescriptor.getMutabilityPlan().disassemble( value, session ) );
+			cacheKey.addHashCode( javaTypeDescriptor.extractHashCode( value ) );
+		}
+		else {
+			final Object relationalValue = converter.toRelationalValue( value );
+			final JavaType relationalJavaType = converter.getRelationalJavaType();
+			cacheKey.addValue( relationalJavaType.getMutabilityPlan().disassemble( relationalValue, session ) );
+			cacheKey.addHashCode( relationalJavaType.extractHashCode( relationalValue ) );
+		}
+	}
+
 	@Override
 	public <X, Y> int forEachDisassembledJdbcValue(
 			Object value,

hibernate-core/src/main/java/org/hibernate/type/CustomType.java

@@ -12,9 +12,11 @@ import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.util.Arrays;
 import java.util.Map;
+import java.util.Objects;
 
 import org.hibernate.HibernateException;
 import org.hibernate.MappingException;
+import org.hibernate.cache.MutableCacheKeyBuilder;
 import org.hibernate.engine.spi.Mapping;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -216,6 +218,33 @@ public class CustomType<J>
 		return disassembled;
 	}
 
+	@Override
+	public void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session) {
+		final Serializable disassembled = getUserType().disassemble( (J) value );
+		// Since UserType#disassemble is an optional operation,
+		// we have to handle the fact that it could produce a null value,
+		// in which case we will try to use a converter for disassembling,
+		// or if that doesn't exist, simply use the domain value as is
+		if ( disassembled == null && value != null ) {
+			final BasicValueConverter<J, Object> valueConverter = getUserType().getValueConverter();
+			if ( valueConverter == null ) {
+				cacheKey.addValue( value );
+			}
+			else {
+				cacheKey.addValue(
+						valueConverter.getRelationalJavaType().getMutabilityPlan().disassemble(
+								valueConverter.toRelationalValue( (J) value ),
+								session
+						)
+				);
+			}
+		}
+		else {
+			cacheKey.addValue( disassembled );
+		}
+		cacheKey.addHashCode( getUserType().hashCode( (J) value ) );
+	}
+
 	@Override
 	public Object replace(
 			Object original,

hibernate-core/src/main/java/org/hibernate/type/descriptor/java/ComponentArrayMutabilityPlan.java

@@ -1,58 +0,0 @@
-/*
- * 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.type.descriptor.java;
-
-import java.io.Serializable;
-import java.util.Arrays;
-
-import org.hibernate.SharedSessionContract;
-
-/**
- * Mutability plan for component based arrays.
- *
- * @author Christian Beikov
- */
-public class ComponentArrayMutabilityPlan implements MutabilityPlan<Object[]> {
-
-	private final JavaType<Object>[] components;
-
-	public ComponentArrayMutabilityPlan(JavaType<Object>[] components) {
-		this.components = components;
-	}
-
-	@Override
-	public boolean isMutable() {
-		return true;
-	}
-
-	@Override
-	public Serializable disassemble(Object[] value, SharedSessionContract session) {
-		return (Serializable) deepCopy( value );
-	}
-
-	@Override
-	public Object[] assemble(Serializable cached, SharedSessionContract session) {
-		return deepCopy( (Object[]) cached );
-	}
-
-	@Override
-	public final Object[] deepCopy(Object[] value) {
-		if ( value == null ) {
-			return null;
-		}
-		if ( value.length != components.length ) {
-			throw new IllegalArgumentException(
-					"Value does not have the expected size " + components.length + ": " + Arrays.toString( value )
-			);
-		}
-		final Object[] copy = new Object[value.length];
-		for ( int i = 0; i < components.length; i++ ) {
-			copy[i] = components[i].getMutabilityPlan().deepCopy( value[i] );
-		}
-		return copy;
-	}
-}

hibernate-core/src/main/java/org/hibernate/type/descriptor/java/JavaType.java

@@ -21,7 +21,6 @@ import org.hibernate.internal.util.ReflectHelper;
 import org.hibernate.internal.util.compare.ComparableComparator;
 import org.hibernate.sql.ast.spi.SqlAppender;
 import org.hibernate.type.descriptor.WrapperOptions;
-import org.hibernate.type.descriptor.java.spi.UnknownBasicJavaType;
 import org.hibernate.type.descriptor.jdbc.JdbcType;
 import org.hibernate.type.descriptor.jdbc.JdbcTypeIndicators;
 import org.hibernate.type.spi.TypeConfiguration;

hibernate-core/src/main/java/org/hibernate/type/descriptor/java/JavaTypedExpressible.java

@@ -6,9 +6,15 @@
  */
 package org.hibernate.type.descriptor.java;
 
+import org.hibernate.cache.MutableCacheKeyBuilder;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
+
 /**
  * Contract for something that has an associated JavaType
  */
 public interface JavaTypedExpressible<T> {
 	JavaType<T> getExpressibleJavaType();
+
+	void addToCacheKey(MutableCacheKeyBuilder cacheKey, Object value, SharedSessionContractImplementor session);
+
 }

hibernate-core/src/main/java/org/hibernate/type/internal/UserTypeJavaTypeWrapper.java

@@ -177,12 +177,43 @@ public class UserTypeJavaTypeWrapper<J> implements BasicJavaType<J> {
 
 		@Override
 		public Serializable disassemble(J value, SharedSessionContract session) {
-			return userType.disassemble( value );
+			final Serializable disassembled = userType.disassemble( value );
+			// Since UserType#disassemble is an optional operation,
+			// we have to handle the fact that it could produce a null value,
+			// in which case we will try to use a converter for disassembling,
+			// or if that doesn't exist, simply use the domain value as is
+			if ( disassembled == null && value != null ) {
+				final BasicValueConverter<J, Object> valueConverter = userType.getValueConverter();
+				if ( valueConverter == null ) {
+					return (Serializable) value;
+				}
+				else {
+					return valueConverter.getRelationalJavaType().getMutabilityPlan().disassemble(
+							valueConverter.toRelationalValue( value ),
+							session
+					);
+				}
+			}
+			return disassembled;
 		}
 
 		@Override
 		public J assemble(Serializable cached, SharedSessionContract session) {
-			return userType.assemble( cached , session);
+			final J assembled = userType.assemble( cached, null );
+			// Since UserType#assemble is an optional operation,
+			// we have to handle the fact that it could produce a null value,
+			// in which case we will try to use a converter for assembling,
+			// or if that doesn't exist, simply use the relational value as is
+			if ( assembled == null && cached != null ) {
+				final BasicValueConverter<J, Object> valueConverter = userType.getValueConverter();
+				if ( valueConverter == null ) {
+					return (J) cached;
+				}
+				else {
+					return valueConverter.toDomainValue( cached );
+				}
+			}
+			return assembled;
 		}
 	}
 }