HHH-14409 : Internal format of natural-id values

At the moment, internally the value of a natural-id is always kept as an array.  For simple natural-ids that means creating an unnecessary array to wrap the simple value.  Change this to allow Object to allow for these simple values

hibernate-core/src/main/java/org/hibernate/action/internal/AbstractEntityInsertAction.java

@@ -17,6 +17,7 @@
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.engine.spi.Status;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.persister.entity.EntityPersister;
 
 /**
@@ -163,13 +164,15 @@ public void afterDeserialize(SharedSessionContractImplementor session) {
 	 */
 	protected void handleNaturalIdPreSaveNotifications() {
 		// before save, we need to add a local (transactional) natural id cross-reference
-		getSession().getPersistenceContextInternal().getNaturalIdHelper().manageLocalNaturalIdCrossReference(
+		final NaturalIdMapping naturalIdMapping = getPersister().getNaturalIdMapping();
-				getPersister(),
+		if ( naturalIdMapping != null ) {
-				getId(),
+			getSession().getPersistenceContextInternal().getNaturalIdHelper().manageLocalResolution(
-				state,
+					getId(),
-				null,
+					naturalIdMapping.extractNaturalIdValues( state, getSession() ),
-				CachedNaturalIdValueSource.INSERT
+					getPersister(),
-		);
+					CachedNaturalIdValueSource.INSERT
+			);
+		}
 	}
 
 	/**
@@ -178,23 +181,29 @@ protected void handleNaturalIdPreSaveNotifications() {
 	 * @param generatedId The generated entity identifier
 	 */
 	public void handleNaturalIdPostSaveNotifications(Object generatedId) {
+		final NaturalIdMapping naturalIdMapping = getPersister().getNaturalIdMapping();
+		if ( naturalIdMapping == null ) {
+			return;
+		}
+
+		final Object naturalIdValues = naturalIdMapping.extractNaturalIdValues( state, getSession() );
+
 		final PersistenceContext.NaturalIdHelper naturalIdHelper = getSession().getPersistenceContextInternal().getNaturalIdHelper();
 		if ( isEarlyInsert() ) {
 			// with early insert, we still need to add a local (transactional) natural id cross-reference
-			naturalIdHelper.manageLocalNaturalIdCrossReference(
+			naturalIdHelper.manageLocalResolution(
-					getPersister(),
 					generatedId,
-					state,
+					naturalIdValues,
-					null,
+					getPersister(),
 					CachedNaturalIdValueSource.INSERT
 			);
 		}
 		// after save, we need to manage the shared cache entries
-		naturalIdHelper.manageSharedNaturalIdCrossReference(
+		naturalIdHelper.manageSharedResolution(
-				getPersister(),
 				generatedId,
-				state,
+				naturalIdValues,
 				null,
+				getPersister(),
 				CachedNaturalIdValueSource.INSERT
 		);
 	}

hibernate-core/src/main/java/org/hibernate/action/internal/EntityDeleteAction.java

@@ -21,6 +21,7 @@
 import org.hibernate.event.spi.PostDeleteEventListener;
 import org.hibernate.event.spi.PreDeleteEvent;
 import org.hibernate.event.spi.PreDeleteEventListener;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.stat.spi.StatisticsImplementor;
 
@@ -33,7 +34,9 @@ public class EntityDeleteAction extends EntityAction {
 	private final Object[] state;
 
 	private SoftLock lock;
-	private Object[] naturalIdValues;
+
+	private final NaturalIdMapping naturalIdMapping;
+	private Object naturalIdValues;
 
 	/**
 	 * Constructs an EntityDeleteAction.
@@ -58,12 +61,15 @@ public EntityDeleteAction(
 		this.isCascadeDeleteEnabled = isCascadeDeleteEnabled;
 		this.state = state;
 
-		// before remove we need to remove the local (transactional) natural id cross-reference
+		this.naturalIdMapping = persister.getNaturalIdMapping();
-		naturalIdValues = session.getPersistenceContextInternal().getNaturalIdHelper().removeLocalNaturalIdCrossReference(
+
-				getPersister(),
+		if ( naturalIdMapping != null ) {
-				getId(),
+			naturalIdValues = session.getPersistenceContextInternal().getNaturalIdHelper().removeLocalResolution(
-				state
+					getPersister(),
-		);
+					getId(),
+					naturalIdMapping.extractNaturalIdValues( state, session )
+			);
+		}
 	}
 
 	public Object getVersion() {
@@ -78,7 +84,7 @@ public Object[] getState() {
 		return state;
 	}
 
-	protected Object[] getNaturalIdValues() {
+	protected Object getNaturalIdValues() {
 		return naturalIdValues;
 	}
 
@@ -139,7 +145,7 @@ public void execute() throws HibernateException {
 			persister.getCacheAccessStrategy().remove( session, ck);
 		}
 
-		persistenceContext.getNaturalIdHelper().removeSharedNaturalIdCrossReference( persister, id, naturalIdValues );
+		persistenceContext.getNaturalIdHelper().removeSharedResolution( persister, id, naturalIdValues );
 
 		postDelete();
 

hibernate-core/src/main/java/org/hibernate/action/internal/EntityUpdateAction.java

@@ -27,6 +27,7 @@
 import org.hibernate.event.spi.PostUpdateEventListener;
 import org.hibernate.event.spi.PreUpdateEvent;
 import org.hibernate.event.spi.PreUpdateEventListener;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.stat.internal.StatsHelper;
 import org.hibernate.stat.spi.StatisticsImplementor;
@@ -42,11 +43,13 @@ public class EntityUpdateAction extends EntityAction {
 	private final int[] dirtyFields;
 	private final boolean hasDirtyCollection;
 	private final Object rowId;
-	private final Object[] previousNaturalIdValues;
 	private Object nextVersion;
 	private Object cacheEntry;
 	private SoftLock lock;
 
+	private final NaturalIdMapping naturalIdMapping;
+	private final Object previousNaturalIdValues;
+
 	/**
 	 * Constructs an EntityUpdateAction
 	 *  @param id The entity identifier
@@ -82,24 +85,23 @@ public EntityUpdateAction(
 		this.hasDirtyCollection = hasDirtyCollection;
 		this.rowId = rowId;
 
-		this.previousNaturalIdValues = determinePreviousNaturalIdValues( persister, id, previousState, session );
+		this.naturalIdMapping = persister.getNaturalIdMapping();
-		session.getPersistenceContextInternal().getNaturalIdHelper().manageLocalNaturalIdCrossReference(
+		if ( naturalIdMapping == null ) {
-				persister,
+			previousNaturalIdValues = null;
-				id,
+		}
-				state,
+		else {
-				previousNaturalIdValues,
+			this.previousNaturalIdValues = determinePreviousNaturalIdValues( persister, id, previousState, session );
-				CachedNaturalIdValueSource.UPDATE
+			session.getPersistenceContextInternal().getNaturalIdHelper().manageLocalResolution(
-		);
+					id, state, persister,
+					CachedNaturalIdValueSource.UPDATE
+			);
+		}
 	}
 
-	private Object[] determinePreviousNaturalIdValues(
+	private static Object determinePreviousNaturalIdValues(
 			EntityPersister persister,
 			Object id, Object[] previousState,
 			SharedSessionContractImplementor session) {
-		if ( ! persister.hasNaturalIdentifier() ) {
-			return null;
-		}
-
 		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
 		if ( previousState != null ) {
 			return persistenceContext.getNaturalIdHelper().extractNaturalIdValues( previousState, persister );
@@ -116,50 +118,10 @@ public Object[] getPreviousState() {
 		return previousState;
 	}
 
-	public Object getPreviousVersion() {
-		return previousVersion;
-	}
-
-	public Object getNextVersion() {
-		return nextVersion;
-	}
-
-	public void setNextVersion(Object nextVersion) {
-		this.nextVersion = nextVersion;
-	}
-
-	public int[] getDirtyFields() {
-		return dirtyFields;
-	}
-
-	public boolean hasDirtyCollection() {
-		return hasDirtyCollection;
-	}
-
 	public Object getRowId() {
 		return rowId;
 	}
 
-	public Object[] getPreviousNaturalIdValues() {
-		return previousNaturalIdValues;
-	}
-
-	protected Object getCacheEntry() {
-		return cacheEntry;
-	}
-
-	protected void setCacheEntry(Object cacheEntry) {
-		this.cacheEntry = cacheEntry;
-	}
-
-	protected SoftLock getLock() {
-		return lock;
-	}
-
-	protected void setLock(SoftLock lock) {
-		this.lock = lock;
-	}
-
 	@Override
 	public void execute() throws HibernateException {
 		final Object id = getId();
@@ -256,13 +218,15 @@ else if ( session.getCacheMode().isPutEnabled() ) {
 			}
 		}
 
-		session.getPersistenceContextInternal().getNaturalIdHelper().manageSharedNaturalIdCrossReference(
+		if ( naturalIdMapping != null ) {
-				persister,
+			session.getPersistenceContextInternal().getNaturalIdHelper().manageSharedResolution(
-				id,
+					id,
-				state,
+					naturalIdMapping.extractNaturalIdValues( state, session ),
-				previousNaturalIdValues,
+					naturalIdMapping.extractNaturalIdValues( previousState, session ),
-				CachedNaturalIdValueSource.UPDATE
+					persister,
-		);
+					CachedNaturalIdValueSource.UPDATE
+			);
+		}
 
 		postUpdate();
 

hibernate-core/src/main/java/org/hibernate/cache/internal/DefaultCacheKeysFactory.java

@@ -50,7 +50,7 @@ public static Object staticCreateEntityKey(Object id, EntityPersister persister,
 		return new CacheKeyImplementation( id, persister.getIdentifierType(), persister.getRootEntityName(), tenantIdentifier, factory );
 	}
 
-	public static Object staticCreateNaturalIdKey(Object[] naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session) {
+	public static Object staticCreateNaturalIdKey(Object naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session) {
 		return new NaturalIdCacheKey( naturalIdValues,  persister.getPropertyTypes(), persister.getNaturalIdentifierProperties(), persister.getRootEntityName(), session );
 	}
 
@@ -62,7 +62,7 @@ public static Object staticGetCollectionId(Object cacheKey) {
 		return ((CacheKeyImplementation) cacheKey).getId();
 	}
 
-	public static Object[] staticGetNaturalIdValues(Object cacheKey) {
+	public static Object staticGetNaturalIdValues(Object cacheKey) {
 		return ((NaturalIdCacheKey) cacheKey).getNaturalIdValues();
 	}
 
@@ -77,7 +77,7 @@ public Object createEntityKey(Object id, EntityPersister persister, SessionFacto
 	}
 
 	@Override
-	public Object createNaturalIdKey(Object[] naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session) {
+	public Object createNaturalIdKey(Object naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session) {
 		return staticCreateNaturalIdKey(naturalIdValues, persister, session);
 	}
 
@@ -92,7 +92,7 @@ public Object getCollectionId(Object cacheKey) {
 	}
 
 	@Override
-	public Object[] getNaturalIdValues(Object cacheKey) {
+	public Object getNaturalIdValues(Object cacheKey) {
 		return staticGetNaturalIdValues(cacheKey);
 	}
 }

hibernate-core/src/main/java/org/hibernate/cache/internal/NaturalIdCacheKey.java

@@ -9,13 +9,12 @@
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.Serializable;
-import java.util.Arrays;
 import java.util.Objects;
 
-import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.ValueHolder;
-import org.hibernate.type.EntityType;
+import org.hibernate.metamodel.mapping.EntityMappingType;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.type.Type;
 
 /**
@@ -28,7 +27,7 @@
  * @author Steve Ebersole
  */
 public class NaturalIdCacheKey implements Serializable {
-	private final Object[] naturalIdValues;
+	private final Object naturalIdValues;
 	private final String entityName;
 	private final String tenantId;
 	private final int hashCode;
@@ -43,40 +42,20 @@ public class NaturalIdCacheKey implements Serializable {
 	 * @param session The originating session
 	 */
 	public NaturalIdCacheKey(
-			final Object[] naturalIdValues,
+			final Object naturalIdValues,
-			Type[] propertyTypes, int[] naturalIdPropertyIndexes, final String entityName,
+			Type[] propertyTypes,
+			int[] naturalIdPropertyIndexes,
+			final String entityName,
 			final SharedSessionContractImplementor session) {
-
 		this.entityName = entityName;
 		this.tenantId = session.getTenantIdentifier();
 
-		this.naturalIdValues = new Serializable[naturalIdValues.length];
+		final EntityMappingType entityMappingType = session.getFactory().getRuntimeMetamodels().getEntityMappingType( entityName );
+		final NaturalIdMapping naturalIdMapping = entityMappingType.getNaturalIdMapping();
 
-		final SessionFactoryImplementor factory = session.getFactory();
+		this.naturalIdValues = naturalIdMapping.disassemble( naturalIdValues, session );
+		this.hashCode = naturalIdMapping.calculateHashCode( naturalIdValues, session );
 
-		final int prime = 31;
-		int result = 1;
-		result = prime * result + ( ( this.entityName == null ) ? 0 : this.entityName.hashCode() );
-		result = prime * result + ( ( this.tenantId == null ) ? 0 : this.tenantId.hashCode() );
-		for ( int i = 0; i < naturalIdValues.length; i++ ) {
-			final int naturalIdPropertyIndex = naturalIdPropertyIndexes[i];
-			final Type type = propertyTypes[naturalIdPropertyIndex];
-			final Object value = naturalIdValues[i];
-
-			result = prime * result + (value != null ? type.getHashCode( value, factory ) : 0);
-
-			// The natural id may not be fully resolved in some situations.  See HHH-7513 for one of them
-			// (re-attaching a mutable natural id uses a database snapshot and hydration does not resolve associations).
-			// TODO: The snapshot should probably be revisited at some point.  Consider semi-resolving, hydrating, etc.
-			if (type instanceof EntityType && type.getSemiResolvedType( factory ).getReturnedClass().isInstance( value )) {
-				this.naturalIdValues[i] = value;
-			}
-			else {
-				this.naturalIdValues[i] = type.disassemble( value, session, null );
-			}
-		}
-
-		this.hashCode = result;
 		initTransients();
 	}
 
@@ -87,15 +66,20 @@ private void initTransients() {
 					public String initialize() {
 						//Complex toString is needed as naturalIds for entities are not simply based on a single value like primary keys
 						//the only same way to differentiate the keys is to include the disassembled values in the string.
-						final StringBuilder toStringBuilder = new StringBuilder().append( entityName ).append(
+						final StringBuilder toStringBuilder = new StringBuilder()
-								"##NaturalId[" );
+								.append( entityName ).append( "##NaturalId[" );
-						for ( int i = 0; i < naturalIdValues.length; i++ ) {
+						if ( naturalIdValues instanceof Object[] ) {
-							toStringBuilder.append( naturalIdValues[i] );
+							final Object[] values = (Object[]) naturalIdValues;
-							if ( i + 1 < naturalIdValues.length ) {
+							for ( int i = 0; i < values.length; i++ ) {
-								toStringBuilder.append( ", " );
+								toStringBuilder.append( values[ i ] );
+								if ( i + 1 < values.length ) {
+									toStringBuilder.append( ", " );
+								}
 							}
 						}
-						toStringBuilder.append( "]" );
+						else {
+							toStringBuilder.append( naturalIdValues );
+						}
 
 						return toStringBuilder.toString();
 					}
@@ -114,7 +98,7 @@ public String getTenantId() {
 	}
 
 	@SuppressWarnings( {"UnusedDeclaration"})
-	public Object[] getNaturalIdValues() {
+	public Object getNaturalIdValues() {
 		return naturalIdValues;
 	}
 
@@ -145,7 +129,7 @@ public boolean equals(Object o) {
 		final NaturalIdCacheKey other = (NaturalIdCacheKey) o;
 		return Objects.equals( entityName, other.entityName )
 				&& Objects.equals( tenantId, other.tenantId )
-				&& Arrays.deepEquals( this.naturalIdValues, other.naturalIdValues );
+				&& Objects.deepEquals( this.naturalIdValues, other.naturalIdValues );
 	}
 
 	private void readObject(ObjectInputStream ois)

hibernate-core/src/main/java/org/hibernate/cache/internal/SimpleCacheKeysFactory.java

@@ -32,7 +32,7 @@ public Object createEntityKey(Object id, EntityPersister persister, SessionFacto
 	}
 
 	@Override
-	public Object createNaturalIdKey(Object[] naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session) {
+	public Object createNaturalIdKey(Object naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session) {
 		// natural ids always need to be wrapped
 		return new NaturalIdCacheKey(naturalIdValues, persister.getPropertyTypes(), persister.getNaturalIdentifierProperties(), null, session);
 	}
@@ -48,7 +48,7 @@ public Object getCollectionId(Object cacheKey) {
 	}
 
 	@Override
-	public Object[] getNaturalIdValues(Object cacheKey) {
+	public Object getNaturalIdValues(Object cacheKey) {
 		return ((NaturalIdCacheKey) cacheKey).getNaturalIdValues();
 	}
 }

hibernate-core/src/main/java/org/hibernate/cache/spi/CacheKeysFactory.java

@@ -19,11 +19,11 @@ public interface CacheKeysFactory {
 
 	Object createEntityKey(Object id, EntityPersister persister, SessionFactoryImplementor factory, String tenantIdentifier);
 
-	Object createNaturalIdKey(Object[] naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session);
+	Object createNaturalIdKey(Object naturalIdValues, EntityPersister persister, SharedSessionContractImplementor session);
 
 	Object getEntityId(Object cacheKey);
 
 	Object getCollectionId(Object cacheKey);
 
-	Object[] getNaturalIdValues(Object cacheKey);
+	Object getNaturalIdValues(Object cacheKey);
 }

hibernate-core/src/main/java/org/hibernate/cache/spi/access/NaturalIdDataAccess.java

@@ -47,7 +47,7 @@ public interface NaturalIdDataAccess extends CachedDomainDataAccess {
 	 * @return a key which can be used to identify an element unequivocally on this same region
 	 */
 	Object generateCacheKey(
-			Object[] naturalIdValues,
+			Object naturalIdValues,
 			EntityPersister rootEntityDescriptor,
 			SharedSessionContractImplementor session);
 
@@ -58,7 +58,7 @@ Object generateCacheKey(
 	 *
 	 * @return the sequence of values which unequivocally identifies a cached element on this region
 	 */
-	Object[] getNaturalIdValues(Object cacheKey);
+	Object getNaturalIdValues(Object cacheKey);
 
 	/**
 	 * Called afterQuery an item has been inserted (beforeQuery the transaction completes),

hibernate-core/src/main/java/org/hibernate/cache/spi/support/AbstractNaturalIdDataAccess.java

@@ -31,14 +31,14 @@ public AbstractNaturalIdDataAccess(
 
 	@Override
 	public Object generateCacheKey(
-			Object[] naturalIdValues,
+			Object naturalIdValues,
 			EntityPersister persister,
 			SharedSessionContractImplementor session) {
 		return keysFactory.createNaturalIdKey( naturalIdValues, persister, session );
 	}
 
 	@Override
-	public Object[] getNaturalIdValues(Object cacheKey) {
+	public Object getNaturalIdValues(Object cacheKey) {
 		return keysFactory.getNaturalIdValues( cacheKey );
 	}
 

hibernate-core/src/main/java/org/hibernate/cache/spi/support/NaturalIdReadWriteAccess.java

@@ -53,14 +53,14 @@ protected Comparator getVersionComparator() {
 
 	@Override
 	public Object generateCacheKey(
-			Object[] naturalIdValues,
+			Object naturalIdValues,
 			EntityPersister rootEntityDescriptor,
 			SharedSessionContractImplementor session) {
 		return keysFactory.createNaturalIdKey( naturalIdValues, rootEntityDescriptor, session );
 	}
 
 	@Override
-	public Object[] getNaturalIdValues(Object cacheKey) {
+	public Object getNaturalIdValues(Object cacheKey) {
 		return keysFactory.getNaturalIdValues( cacheKey );
 	}
 

hibernate-core/src/main/java/org/hibernate/engine/internal/AbstractEntityEntry.java

@@ -408,11 +408,8 @@ public void setReadOnly(boolean readOnly, Object entity) {
 			}
 			setStatus( Status.MANAGED );
 			loadedState = getPersister().getPropertyValues( entity );
-			getPersistenceContext().getNaturalIdHelper().manageLocalNaturalIdCrossReference(
+			getPersistenceContext().getNaturalIdHelper().manageLocalResolution(
-					persister,
+					id, loadedState, persister,
-					id,
-					loadedState,
-					null,
 					CachedNaturalIdValueSource.LOAD
 			);
 		}

hibernate-core/src/main/java/org/hibernate/engine/internal/NaturalIdXrefDelegate.java

@@ -8,7 +8,6 @@
 
 import java.io.Serializable;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -19,10 +18,10 @@
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.stat.internal.StatsHelper;
 import org.hibernate.stat.spi.StatisticsImplementor;
-import org.hibernate.type.Type;
 
 import org.jboss.logging.Logger;
 
@@ -70,12 +69,12 @@ protected SharedSessionContractImplementor session() {
 	 *
 	 * @return {@code true} if a new entry was actually added; {@code false} otherwise.
 	 */
-	public boolean cacheNaturalIdCrossReference(EntityPersister persister, Object pk, Object[] naturalIdValues) {
+	public boolean cacheResolution(EntityPersister persister, Object pk, Object naturalIdValues) {
 		validateNaturalId( persister, naturalIdValues );
 
 		NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
 		if ( entityNaturalIdResolutionCache == null ) {
-			entityNaturalIdResolutionCache = new NaturalIdResolutionCache( persister );
+			entityNaturalIdResolutionCache = new NaturalIdResolutionCache( persister, persistenceContext );
 			NaturalIdResolutionCache previousInstance = naturalIdResolutionCacheMap.putIfAbsent( persister, entityNaturalIdResolutionCache );
 			if ( previousInstance != null ) {
 				entityNaturalIdResolutionCache = previousInstance;
@@ -93,29 +92,29 @@ public boolean cacheNaturalIdCrossReference(EntityPersister persister, Object pk
 	 *
 	 * @return The cached values, if any.  May be different from incoming values.
 	 */
-	public Object[] removeNaturalIdCrossReference(EntityPersister persister, Object pk, Object[] naturalIdValues) {
+	public Object removeResolutions(EntityPersister persister, Object pk, Object naturalIdValues) {
 		persister = locatePersisterForKey( persister );
+
+		final NaturalIdMapping naturalIdMapping = persister.getNaturalIdMapping();
 		validateNaturalId( persister, naturalIdValues );
 
 		final NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
-		Object[] sessionCachedNaturalIdValues = null;
+		Object sessionCachedNaturalIdValues = null;
 		if ( entityNaturalIdResolutionCache != null ) {
-			final CachedNaturalId cachedNaturalId = entityNaturalIdResolutionCache.pkToNaturalIdMap
+			final CachedNaturalId cachedNaturalId = entityNaturalIdResolutionCache.pkToNaturalIdMap.remove( pk );
-					.remove( pk );
 			if ( cachedNaturalId != null ) {
 				entityNaturalIdResolutionCache.naturalIdToPkMap.remove( cachedNaturalId );
-				sessionCachedNaturalIdValues = cachedNaturalId.getValues();
+				sessionCachedNaturalIdValues = cachedNaturalId.getNaturalId();
 			}
 		}
 
 		if ( persister.hasNaturalIdCache() ) {
-			final NaturalIdDataAccess naturalIdCacheAccessStrategy = persister
+			final NaturalIdDataAccess naturalIdCacheAccessStrategy = persister.getNaturalIdCacheAccessStrategy();
-					.getNaturalIdCacheAccessStrategy();
 			final Object naturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( naturalIdValues, persister, session() );
 			naturalIdCacheAccessStrategy.evict( naturalIdCacheKey );
 
 			if ( sessionCachedNaturalIdValues != null
-					&& !Arrays.equals( sessionCachedNaturalIdValues, naturalIdValues ) ) {
+					&& ! naturalIdMapping.areEqual( sessionCachedNaturalIdValues, naturalIdValues, session() ) ) {
 				final Object sessionNaturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( sessionCachedNaturalIdValues, persister, session() );
 				naturalIdCacheAccessStrategy.evict( sessionNaturalIdCacheKey );
 			}
@@ -133,7 +132,7 @@ public Object[] removeNaturalIdCrossReference(EntityPersister persister, Object
 	 *
 	 * @return {@code true} if the given naturalIdValues match the current cached values; {@code false} otherwise.
 	 */
-	public boolean sameAsCached(EntityPersister persister, Object pk, Object[] naturalIdValues) {
+	public boolean sameAsCached(EntityPersister persister, Object pk, Object naturalIdValues) {
 		final NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
 		return entityNaturalIdResolutionCache != null
 				&& entityNaturalIdResolutionCache.sameAsCached( pk, naturalIdValues );
@@ -157,16 +156,17 @@ protected EntityPersister locatePersisterForKey(EntityPersister persister) {
 	 *     <li>the number of natural id values matches the expected number</li>
 	 * </ul>
 	 *
-	 * @param persister The persister representing the entity type.
+	 * @param entityDescriptor The entity type descriptor
 	 * @param naturalIdValues The natural id values
 	 */
-	protected void validateNaturalId(EntityPersister persister, Object[] naturalIdValues) {
+	protected void validateNaturalId(EntityPersister entityDescriptor, Object naturalIdValues) {
-		if ( !persister.hasNaturalIdentifier() ) {
+		final NaturalIdMapping naturalIdMapping = entityDescriptor.getNaturalIdMapping();
+
+		if ( naturalIdMapping == null ) {
 			throw new IllegalArgumentException( "Entity did not define a natural-id" );
 		}
-		if ( persister.getNaturalIdentifierProperties().length != naturalIdValues.length ) {
+
-			throw new IllegalArgumentException( "Mismatch between expected number of natural-id values and found." );
+		naturalIdMapping.validateInternalForm( naturalIdValues, persistenceContext.getSession() );
-		}
 	}
 
 	/**
@@ -177,7 +177,7 @@ protected void validateNaturalId(EntityPersister persister, Object[] naturalIdVa
 	 *
 	 * @return The corresponding cross-referenced natural id values, or {@code null} if none
 	 */
-	public Object[] findCachedNaturalId(EntityPersister persister, Object pk) {
+	public Object findCachedNaturalId(EntityPersister persister, Object pk) {
 		persister = locatePersisterForKey( persister );
 		final NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
 		if ( entityNaturalIdResolutionCache == null ) {
@@ -189,7 +189,7 @@ public Object[] findCachedNaturalId(EntityPersister persister, Object pk) {
 			return null;
 		}
 
-		return cachedNaturalId.getValues();
+		return cachedNaturalId.getNaturalId();
 	}
 
 	/**
@@ -199,19 +199,19 @@ public Object[] findCachedNaturalId(EntityPersister persister, Object pk) {
 	 *
 	 * @param persister The persister representing the entity type.
 	 * @param naturalIdValues The natural id value(s)
-	 * 
+	 *
-	 * @return The corresponding cross-referenced primary key, 
+	 * @return The corresponding cross-referenced primary key,
 	 * 		{@link PersistenceContext.NaturalIdHelper#INVALID_NATURAL_ID_REFERENCE},
 	 * 		or {@code null} if none 
 	 */
-	public Object findCachedNaturalIdResolution(EntityPersister persister, Object[] naturalIdValues) {
+	public Object findResolution(EntityPersister persister, Object naturalIdValues) {
 		persister = locatePersisterForKey( persister );
 		validateNaturalId( persister, naturalIdValues );
 
 		NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
 
 		Object pk;
-		final CachedNaturalId cachedNaturalId = new CachedNaturalId( persister, naturalIdValues );
+		final CachedNaturalId cachedNaturalId = new CachedNaturalId( naturalIdValues, persister, persistenceContext );
 		if ( entityNaturalIdResolutionCache != null ) {
 			pk = entityNaturalIdResolutionCache.naturalIdToPkMap.get( cachedNaturalId );
 
@@ -221,7 +221,7 @@ public Object findCachedNaturalIdResolution(EntityPersister persister, Object[]
 					LOG.trace(
 							"Resolved natural key -> primary key resolution in session cache: " +
 									persister.getRootEntityName() + "#[" +
-									Arrays.toString( naturalIdValues ) + "]"
+									naturalIdValues + "]"
 					);
 				}
 
@@ -264,14 +264,14 @@ public Object findCachedNaturalIdResolution(EntityPersister persister, Object[]
 				// protected to avoid Arrays.toString call unless needed
 				LOG.tracef(
 						"Found natural key [%s] -> primary key [%s] xref in second-level cache for %s",
-						Arrays.toString( naturalIdValues ),
+						naturalIdValues,
 						pk,
 						persister.getRootEntityName()
 				);
 			}
 
 			if ( entityNaturalIdResolutionCache == null ) {
-				entityNaturalIdResolutionCache = new NaturalIdResolutionCache( persister );
+				entityNaturalIdResolutionCache = new NaturalIdResolutionCache( persister, persistenceContext );
 				NaturalIdResolutionCache existingCache = naturalIdResolutionCacheMap.putIfAbsent( persister, entityNaturalIdResolutionCache );
 				if ( existingCache != null ) {
 					entityNaturalIdResolutionCache = existingCache;
@@ -329,7 +329,7 @@ public Collection<Object> getCachedPkResolutions(EntityPersister persister) {
 	 *
 	 * @see org.hibernate.NaturalIdLoadAccess#setSynchronizationEnabled
 	 */
-	public void stashInvalidNaturalIdReference(EntityPersister persister, Object[] invalidNaturalIdValues) {
+	public void stashInvalidNaturalIdReference(EntityPersister persister, Object invalidNaturalIdValues) {
 		persister = locatePersisterForKey( persister );
 
 		final NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
@@ -354,35 +354,27 @@ public void unStashInvalidNaturalIdReferences() {
 	 * Used to put natural id values into collections.  Useful mainly to apply equals/hashCode implementations.
 	 */
 	private static class CachedNaturalId implements Serializable {
+		private final PersistenceContext persistenceContext;
 		private final EntityPersister persister;
-		private final Object[] values;
+		private final Object naturalId;
-		private final Type[] naturalIdTypes;
 		private int hashCode;
 
-		public CachedNaturalId(EntityPersister persister, Object[] values) {
+		public CachedNaturalId(Object naturalIdValue, EntityPersister persister, PersistenceContext persistenceContext) {
+			this.persistenceContext = persistenceContext;
+
 			this.persister = persister;
-			this.values = values;
+			this.naturalId = naturalIdValue;
 
 			final int prime = 31;
 			int hashCodeCalculation = 1;
 			hashCodeCalculation = prime * hashCodeCalculation + persister.hashCode();
-
+			hashCodeCalculation = prime * hashCodeCalculation + persister.getNaturalIdMapping().calculateHashCode( naturalIdValue, persistenceContext.getSession());
-			final int[] naturalIdPropertyIndexes = persister.getNaturalIdentifierProperties();
-			naturalIdTypes = new Type[ naturalIdPropertyIndexes.length ];
-			int i = 0;
-			for ( int naturalIdPropertyIndex : naturalIdPropertyIndexes ) {
-				final Type type = persister.getPropertyType( persister.getPropertyNames()[ naturalIdPropertyIndex ] );
-				naturalIdTypes[i] = type;
-				final int elementHashCode = values[i] == null ? 0 :type.getHashCode( values[i], persister.getFactory() );
-				hashCodeCalculation = prime * hashCodeCalculation + elementHashCode;
-				i++;
-			}
 
 			this.hashCode = hashCodeCalculation;
 		}
 
-		public Object[] getValues() {
+		public Object getNaturalId() {
-			return values;
+			return naturalId;
 		}
 
 		@Override
@@ -403,17 +395,11 @@ public boolean equals(Object obj) {
 			}
 
 			final CachedNaturalId other = (CachedNaturalId) obj;
-			return persister.equals( other.persister ) && isSame( other.values );
+			return persister.equals( other.persister ) && isSame( other.naturalId );
 		}
 
-		private boolean isSame(Object[] otherValues) {
+		private boolean isSame(Object otherValue) {
-			// lengths have already been verified at this point
+			return persister.getNaturalIdMapping().areEqual( naturalId, otherValue, persistenceContext.getSession() );
-			for ( int i = 0; i < naturalIdTypes.length; i++ ) {
-				if ( ! naturalIdTypes[i].isEqual( values[i], otherValues[i], persister.getFactory() ) ) {
-					return false;
-				}
-			}
-			return true;
 		}
 	}
 
@@ -421,6 +407,8 @@ private boolean isSame(Object[] otherValues) {
 	 * Represents the persister-specific cross-reference cache.
 	 */
 	private static class NaturalIdResolutionCache implements Serializable {
+		private final PersistenceContext persistenceContext;
+
 		private final EntityPersister persister;
 
 		private Map<Object, CachedNaturalId> pkToNaturalIdMap = new ConcurrentHashMap<>();
@@ -428,15 +416,16 @@ private static class NaturalIdResolutionCache implements Serializable {
 
 		private List<CachedNaturalId> invalidNaturalIdList;
 
-		private NaturalIdResolutionCache(EntityPersister persister) {
+		private NaturalIdResolutionCache(EntityPersister persister, PersistenceContext persistenceContext) {
 			this.persister = persister;
+			this.persistenceContext = persistenceContext;
 		}
 
 		public EntityPersister getPersister() {
 			return persister;
 		}
 
-		public boolean sameAsCached(Object pk, Object[] naturalIdValues) {
+		public boolean sameAsCached(Object pk, Object naturalIdValues) {
 			if ( pk == null ) {
 				return false;
 			}
@@ -449,7 +438,7 @@ public boolean sameAsCached(Object pk, Object[] naturalIdValues) {
 			return false;
 		}
 
-		public boolean cache(Object pk, Object[] naturalIdValues) {
+		public boolean cache(Object pk, Object naturalIdValues) {
 			if ( pk == null ) {
 				return false;
 			}
@@ -461,23 +450,23 @@ public boolean cache(Object pk, Object[] naturalIdValues) {
 				naturalIdToPkMap.remove( initial );
 			}
 
-			final CachedNaturalId cachedNaturalId = new CachedNaturalId( persister, naturalIdValues );
+			final CachedNaturalId cachedNaturalId = new CachedNaturalId( naturalIdValues, persister, persistenceContext );
 			pkToNaturalIdMap.put( pk, cachedNaturalId );
 			naturalIdToPkMap.put( cachedNaturalId, pk );
 			
 			return true;
 		}
 
-		public void stashInvalidNaturalIdReference(Object[] invalidNaturalIdValues) {
+		public void stashInvalidNaturalIdReference(Object invalidNaturalIdValues) {
 			if ( invalidNaturalIdList == null ) {
 				invalidNaturalIdList = new ArrayList<>();
 			}
-			invalidNaturalIdList.add( new CachedNaturalId( persister, invalidNaturalIdValues ) );
+			invalidNaturalIdList.add( new CachedNaturalId( invalidNaturalIdValues, persister, persistenceContext ) );
 		}
 
-		public boolean containsInvalidNaturalIdReference(Object[] naturalIdValues) {
+		public boolean containsInvalidNaturalIdReference(Object naturalIdValues) {
 			return invalidNaturalIdList != null
-					&& invalidNaturalIdList.contains( new CachedNaturalId( persister, naturalIdValues ) );
+					&& invalidNaturalIdList.contains( new CachedNaturalId( naturalIdValues, persister, persistenceContext ) );
 		}
 
 		public void unStashInvalidNaturalIdReferences() {

hibernate-core/src/main/java/org/hibernate/engine/internal/StatefulPersistenceContext.java

@@ -61,6 +61,7 @@
 import org.hibernate.internal.util.collections.CollectionHelper;
 import org.hibernate.internal.util.collections.ConcurrentReferenceHashMap;
 import org.hibernate.internal.util.collections.IdentityMap;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.metamodel.spi.MetamodelImplementor;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.persister.entity.EntityPersister;
@@ -328,7 +329,7 @@ public Object[] getDatabaseSnapshot(Object id, EntityPersister persister) throws
 	}
 
 	@Override
-	public Object[] getNaturalIdSnapshot(Object id, EntityPersister persister) throws HibernateException {
+	public Object getNaturalIdSnapshot(Object id, EntityPersister persister) throws HibernateException {
 		if ( !persister.hasNaturalIdentifier() ) {
 			return null;
 		}
@@ -336,7 +337,7 @@ public Object[] getNaturalIdSnapshot(Object id, EntityPersister persister) throw
 		persister = locateProperPersister( persister );
 
 		// let's first see if it is part of the natural id cache...
-		final Object[] cachedValue = naturalIdHelper.findCachedNaturalId( persister, id );
+		final Object cachedValue = naturalIdHelper.findCachedNaturalId( persister, id );
 		if ( cachedValue != null ) {
 			return cachedValue;
 		}
@@ -345,7 +346,7 @@ public Object[] getNaturalIdSnapshot(Object id, EntityPersister persister) throw
 		if ( persister.getEntityMetamodel().hasImmutableNaturalId() ) {
 			// an immutable natural-id is not retrieved during a normal database-snapshot operation...
 			final Object[] dbValue = persister.getNaturalIdentifierSnapshot( id, session );
-			naturalIdHelper.cacheNaturalIdCrossReferenceFromLoad(
+			naturalIdHelper.cacheResolutionFromLoad(
 					persister,
 					id,
 					dbValue
@@ -365,7 +366,7 @@ public Object[] getNaturalIdSnapshot(Object id, EntityPersister persister) throw
 			for ( int i = 0; i < props.length; i++ ) {
 				naturalIdSnapshotSubSet[i] = entitySnapshot[ props[i] ];
 			}
-			naturalIdHelper.cacheNaturalIdCrossReferenceFromLoad(
+			naturalIdHelper.cacheResolutionFromLoad(
 					persister,
 					id,
 					naturalIdSnapshotSubSet
@@ -1947,10 +1948,10 @@ private NaturalIdXrefDelegate getNaturalIdXrefDelegate() {
 
 	private final NaturalIdHelper naturalIdHelper = new NaturalIdHelper() {
 		@Override
-		public void cacheNaturalIdCrossReferenceFromLoad(
+		public void cacheResolutionFromLoad(
 				EntityPersister persister,
 				Object id,
-				Object[] naturalIdValues) {
+				Object naturalIdValues) {
 			if ( !persister.hasNaturalIdentifier() ) {
 				// nothing to do
 				return;
@@ -1962,19 +1963,18 @@ public void cacheNaturalIdCrossReferenceFromLoad(
 			//	from a single load event.  The first put journal would come from the natural id resolution;
 			// the second comes from the entity loading.  In this condition, we want to avoid the multiple
 			// 'put' stats incrementing.
-			final boolean justAddedLocally = getNaturalIdXrefDelegate().cacheNaturalIdCrossReference( persister, id, naturalIdValues );
+			final boolean justAddedLocally = getNaturalIdXrefDelegate().cacheResolution( persister, id, naturalIdValues );
 
 			if ( justAddedLocally && persister.hasNaturalIdCache() ) {
-				managedSharedCacheEntries( persister, id, naturalIdValues, null, CachedNaturalIdValueSource.LOAD );
+				managedSharedResolutions( persister, id, naturalIdValues, null, CachedNaturalIdValueSource.LOAD );
 			}
 		}
 
 		@Override
-		public void manageLocalNaturalIdCrossReference(
+		public void manageLocalResolution(
-				EntityPersister persister,
 				Object id,
-				Object[] state,
+				Object naturalId,
-				Object[] previousState,
+				EntityPersister persister,
 				CachedNaturalIdValueSource source) {
 			if ( !persister.hasNaturalIdentifier() ) {
 				// nothing to do
@@ -1982,18 +1982,17 @@ public void manageLocalNaturalIdCrossReference(
 			}
 
 			persister = locateProperPersister( persister );
-			final Object[] naturalIdValues = extractNaturalIdValues( state, persister );
 
 			// cache
-			getNaturalIdXrefDelegate().cacheNaturalIdCrossReference( persister, id, naturalIdValues );
+			getNaturalIdXrefDelegate().cacheResolution( persister, id, naturalId );
 		}
 
 		@Override
-		public void manageSharedNaturalIdCrossReference(
+		public void manageSharedResolution(
-				EntityPersister persister,
 				final Object id,
-				Object[] state,
+				Object naturalId,
-				Object[] previousState,
+				Object previousNaturalId,
+				EntityPersister persister,
 				CachedNaturalIdValueSource source) {
 			if ( !persister.hasNaturalIdentifier() ) {
 				// nothing to do
@@ -2006,17 +2005,17 @@ public void manageSharedNaturalIdCrossReference(
 			}
 
 			persister = locateProperPersister( persister );
-			final Object[] naturalIdValues = extractNaturalIdValues( state, persister );
+			final Object naturalIdValues = extractNaturalIdValues( naturalId, persister );
-			final Object[] previousNaturalIdValues = previousState == null ? null : extractNaturalIdValues( previousState, persister );
+			final Object previousNaturalIdValues = previousNaturalId == null ? null : extractNaturalIdValues( previousNaturalId, persister );
 
-			managedSharedCacheEntries( persister, id, naturalIdValues, previousNaturalIdValues, source );
+			managedSharedResolutions( persister, id, naturalIdValues, previousNaturalIdValues, source );
 		}
 
-		private void managedSharedCacheEntries(
+		private void managedSharedResolutions(
 				EntityPersister persister,
 				final Object id,
-				Object[] naturalIdValues,
+				Object naturalIdValues,
-				Object[] previousNaturalIdValues,
+				Object previousNaturalIdValues,
 				CachedNaturalIdValueSource source) {
 			final NaturalIdDataAccess naturalIdCacheAccessStrategy = persister.getNaturalIdCacheAccessStrategy();
 			final Object naturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( naturalIdValues, persister, session );
@@ -2133,26 +2132,25 @@ public void doAfterTransactionCompletion(boolean success, SharedSessionContractI
 		}
 
 		@Override
-		public Object[] removeLocalNaturalIdCrossReference(EntityPersister persister, Object id, Object[] state) {
+		public Object removeLocalResolution(EntityPersister persister, Object id, Object naturalId) {
 			if ( !persister.hasNaturalIdentifier() ) {
 				// nothing to do
 				return null;
 			}
 
 			persister = locateProperPersister( persister );
-			final Object[] naturalIdValues = getNaturalIdValues( state, persister );
 
-			final Object[] localNaturalIdValues = getNaturalIdXrefDelegate().removeNaturalIdCrossReference(
+			final Object localNaturalIdValues = getNaturalIdXrefDelegate().removeResolutions(
 					persister, 
-					id, 
+					id,
-					naturalIdValues 
+					naturalId
 			);
 
-			return localNaturalIdValues != null ? localNaturalIdValues : naturalIdValues;
+			return localNaturalIdValues != null ? localNaturalIdValues : naturalId;
 		}
 
 		@Override
-		public void removeSharedNaturalIdCrossReference(EntityPersister persister, Object id, Object[] naturalIdValues) {
+		public void removeSharedResolution(EntityPersister persister, Object id, Object naturalIdValues) {
 			if ( !persister.hasNaturalIdentifier() ) {
 				// nothing to do
 				return;
@@ -2180,46 +2178,37 @@ public void removeSharedNaturalIdCrossReference(EntityPersister persister, Objec
 		}
 
 		@Override
-		public Object[] findCachedNaturalId(EntityPersister persister, Object pk) {
+		public Object findCachedNaturalId(EntityPersister persister, Object pk) {
 			return getNaturalIdXrefDelegate().findCachedNaturalId( locateProperPersister( persister ), pk );
 		}
 
 		@Override
 		public Object findCachedNaturalIdResolution(EntityPersister persister, Object[] naturalIdValues) {
-			return getNaturalIdXrefDelegate().findCachedNaturalIdResolution( locateProperPersister( persister ), naturalIdValues );
+			return getNaturalIdXrefDelegate().findResolution( locateProperPersister( persister ), naturalIdValues );
 		}
 
 		@Override
-		public Object[] extractNaturalIdValues(Object[] state, EntityPersister persister) {
+		public Object extractNaturalIdValues(Object[] state, EntityPersister persister) {
-			final int[] naturalIdPropertyIndexes = persister.getNaturalIdentifierProperties();
+			final NaturalIdMapping naturalIdMapping = persister.getNaturalIdMapping();
-			if ( state.length == naturalIdPropertyIndexes.length ) {
+			assert naturalIdMapping != null;
-				return state;
-			}
 
-			final Object[] naturalIdValues = new Object[naturalIdPropertyIndexes.length];
+			return naturalIdMapping.extractNaturalIdValues( state, getSession() );
-			for ( int i = 0; i < naturalIdPropertyIndexes.length; i++ ) {
-				naturalIdValues[i] = state[naturalIdPropertyIndexes[i]];
-			}
-			return naturalIdValues;
 		}
 
 		@Override
-		public Object[] extractNaturalIdValues(Object entity, EntityPersister persister) {
+		public Object extractNaturalIdValues(Object entity, EntityPersister persister) {
 			if ( entity == null ) {
 				throw new AssertionFailure( "Entity from which to extract natural id value(s) cannot be null" );
 			}
+
 			if ( persister == null ) {
 				throw new AssertionFailure( "Persister to use in extracting natural id value(s) cannot be null" );
 			}
 
-			final int[] naturalIdentifierProperties = persister.getNaturalIdentifierProperties();
+			final NaturalIdMapping naturalIdMapping = persister.getNaturalIdMapping();
-			final Object[] naturalIdValues = new Object[naturalIdentifierProperties.length];
+			assert naturalIdMapping != null;
 
-			for ( int i = 0; i < naturalIdentifierProperties.length; i++ ) {
+			return naturalIdMapping.extractNaturalIdValues( entity, session );
-				naturalIdValues[i] = persister.getPropertyValue( entity, naturalIdentifierProperties[i] );
-			}
-
-			return naturalIdValues;
 		}
 
 		@Override
@@ -2236,7 +2225,7 @@ public void handleSynchronization(EntityPersister persister, Object pk, Object e
 
 			persister = locateProperPersister( persister );
 
-			final Object[] naturalIdValuesFromCurrentObjectState = extractNaturalIdValues( entity, persister );
+			final Object naturalIdValuesFromCurrentObjectState = extractNaturalIdValues( entity, persister );
 			final NaturalIdXrefDelegate naturalIdXrefDelegate = getNaturalIdXrefDelegate();
 			final boolean changed = ! naturalIdXrefDelegate.sameAsCached(
 					persister,
@@ -2245,11 +2234,11 @@ public void handleSynchronization(EntityPersister persister, Object pk, Object e
 			);
 
 			if ( changed ) {
-				final Object[] cachedNaturalIdValues = naturalIdXrefDelegate.findCachedNaturalId( persister, pk );
+				final Object cachedNaturalIdValues = naturalIdXrefDelegate.findCachedNaturalId( persister, pk );
-				naturalIdXrefDelegate.cacheNaturalIdCrossReference( persister, pk, naturalIdValuesFromCurrentObjectState );
+				naturalIdXrefDelegate.cacheResolution( persister, pk, naturalIdValuesFromCurrentObjectState );
 				naturalIdXrefDelegate.stashInvalidNaturalIdReference( persister, cachedNaturalIdValues );
 
-				removeSharedNaturalIdCrossReference(
+				removeSharedResolution(
 						persister,
 						pk,
 						cachedNaturalIdValues
@@ -2264,7 +2253,7 @@ public void cleanupFromSynchronizations() {
 
 		@Override
 		public void handleEviction(Object object, EntityPersister persister, Serializable identifier) {
-			getNaturalIdXrefDelegate().removeNaturalIdCrossReference(
+			getNaturalIdXrefDelegate().removeResolutions(
 					persister,
 					identifier,
 					findCachedNaturalId( persister, identifier )
@@ -2276,15 +2265,4 @@ public void handleEviction(Object object, EntityPersister persister, Serializabl
 	public NaturalIdHelper getNaturalIdHelper() {
 		return naturalIdHelper;
 	}
-
-	private Object[] getNaturalIdValues(Object[] state, EntityPersister persister) {
-		final int[] naturalIdPropertyIndexes = persister.getNaturalIdentifierProperties();
-		final Object[] naturalIdValues = new Object[naturalIdPropertyIndexes.length];
-
-		for ( int i = 0; i < naturalIdPropertyIndexes.length; i++ ) {
-			naturalIdValues[i] = state[naturalIdPropertyIndexes[i]];
-		}
-
-		return naturalIdValues;
-	}
 }

hibernate-core/src/main/java/org/hibernate/engine/internal/TwoPhaseLoad.java

@@ -335,7 +335,7 @@ public static void initializeEntityFromEntityEntryLoadedState(
 		}
 
 		if ( persister.hasNaturalIdentifier() ) {
-			persistenceContext.getNaturalIdHelper().cacheNaturalIdCrossReferenceFromLoad(
+			persistenceContext.getNaturalIdHelper().cacheResolutionFromLoad(
 					persister,
 					id,
 					persistenceContext.getNaturalIdHelper().extractNaturalIdValues( hydratedState, persister )

hibernate-core/src/main/java/org/hibernate/engine/spi/PersistenceContext.java

@@ -147,7 +147,7 @@ public interface PersistenceContext {
 	 *
 	 * @return The current (non-cached) snapshot of the entity's natural id state.
 	 */
-	Object[] getNaturalIdSnapshot(Object id, EntityPersister persister);
+	Object getNaturalIdSnapshot(Object id, EntityPersister persister);
 
 	/**
 	 * Add a canonical mapping from entity key to entity instance
@@ -814,7 +814,7 @@ interface NaturalIdHelper {
 		 *
 		 * @return The extracted natural id values
 		 */
-		Object[] extractNaturalIdValues(Object[] state, EntityPersister persister);
+		Object extractNaturalIdValues(Object[] state, EntityPersister persister);
 
 		/**
 		 * Given an entity instance, extract the values that represent the natural id
@@ -824,35 +824,27 @@ interface NaturalIdHelper {
 		 *
 		 * @return The extracted natural id values
 		 */
-		Object[] extractNaturalIdValues(Object entity, EntityPersister persister);
+		Object extractNaturalIdValues(Object entity, EntityPersister persister);
 
 		/**
 		 * Performs processing related to creating natural-id cross-reference entries on load.
 		 * Handles both the local (transactional) and shared (second-level) caches.
-		 *  @param persister The persister representing the entity type.
+		 * @param persister The persister representing the entity type.
 		 * @param id The primary key value
 		 * @param naturalIdValues The natural id values
 		 */
-		void cacheNaturalIdCrossReferenceFromLoad(
+		void cacheResolutionFromLoad(
 				EntityPersister persister,
 				Object id,
-				Object[] naturalIdValues);
+				Object naturalIdValues);
 
 		/**
 		 * Creates necessary local cross-reference entries.
-		 *
-		 * @param persister The persister representing the entity type.
-		 * @param id The primary key value
-		 * @param state Generally the "full entity state array", though could also be the natural id values array
-		 * @param previousState Generally the "full entity state array", though could also be the natural id values array.
-		 * 		Specifically represents the previous values on update, and so is only used with {@link CachedNaturalIdValueSource#UPDATE}
-		 * @param source Enumeration representing how these values are coming into cache.
 		 */
-		void manageLocalNaturalIdCrossReference(
+		void manageLocalResolution(
-				EntityPersister persister,
 				Object id,
-				Object[] state,
+				Object naturalId,
-				Object[] previousState,
+				EntityPersister persister,
 				CachedNaturalIdValueSource source);
 
 		/**
@@ -860,37 +852,29 @@ void manageLocalNaturalIdCrossReference(
 		 *
 		 * @param persister The persister representing the entity type.
 		 * @param id The primary key value
-		 * @param state Generally the "full entity state array", though could also be the natural id values array
+		 * @param naturalId Generally the "full entity state array", though could also be the natural id values array
 		 *
 		 * @return The local cached natural id values (could be different from given values).
 		 */
-		Object[] removeLocalNaturalIdCrossReference(EntityPersister persister, Object id, Object[] state);
+		Object removeLocalResolution(EntityPersister persister, Object id, Object naturalId);
 
 		/**
 		 * Creates necessary shared (second level cache) cross-reference entries.
-		 *
-		 * @param persister The persister representing the entity type.
-		 * @param id The primary key value
-		 * @param state Generally the "full entity state array", though could also be the natural id values array
-		 * @param previousState Generally the "full entity state array", though could also be the natural id values array.
-		 * 		Specifically represents the previous values on update, and so is only used with {@link CachedNaturalIdValueSource#UPDATE}
-		 * @param source Enumeration representing how these values are coming into cache.
 		 */
-		void manageSharedNaturalIdCrossReference(
+		void manageSharedResolution(
-				EntityPersister persister,
 				Object id,
-				Object[] state,
+				Object naturalId,
-				Object[] previousState,
+				Object previousNaturalId,
+				EntityPersister persister,
 				CachedNaturalIdValueSource source);
 
 		/**
 		 * Cleans up local cross-reference entries.
-		 *
+		 *  @param persister The persister representing the entity type.
-		 * @param persister The persister representing the entity type.
 		 * @param id The primary key value
 		 * @param naturalIdValues The natural id values array
 		 */
-		void removeSharedNaturalIdCrossReference(EntityPersister persister, Object id, Object[] naturalIdValues);
+		void removeSharedResolution(EntityPersister persister, Object id, Object naturalIdValues);
 
 		/**
 		 * Given a persister and primary key, find the corresponding cross-referenced natural id values.
@@ -900,7 +884,7 @@ void manageSharedNaturalIdCrossReference(
 		 *
 		 * @return The cross-referenced natural-id values, or {@code null}
 		 */
-		Object[] findCachedNaturalId(EntityPersister persister, Object pk);
+		Object findCachedNaturalId(EntityPersister persister, Object pk);
 
 		/**
 		 * Given a persister and natural-id values, find the corresponding cross-referenced primary key. Will return

hibernate-core/src/main/java/org/hibernate/event/internal/DefaultFlushEntityEventListener.java

@@ -22,7 +22,6 @@
 import org.hibernate.engine.internal.Versioning;
 import org.hibernate.engine.spi.EntityEntry;
 import org.hibernate.engine.spi.EntityKey;
-import org.hibernate.engine.spi.ManagedEntity;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.PersistentAttributeInterceptable;
 import org.hibernate.engine.spi.PersistentAttributeInterceptor;
@@ -38,9 +37,9 @@
 import org.hibernate.jpa.event.spi.CallbackRegistry;
 import org.hibernate.jpa.event.spi.CallbackRegistryConsumer;
 import org.hibernate.metadata.ClassMetadata;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.pretty.MessageHelper;
-import org.hibernate.proxy.HibernateProxy;
 import org.hibernate.stat.spi.StatisticsImplementor;
 import org.hibernate.type.Type;
 
@@ -103,48 +102,17 @@ private void checkNaturalId(
 			}
 		}
 
-		if ( persister.hasNaturalIdentifier() && entry.getStatus() != Status.READ_ONLY ) {
+		final NaturalIdMapping naturalIdMapping = persister.getNaturalIdMapping();
-			if ( !persister.getEntityMetamodel().hasImmutableNaturalId() ) {
+		if ( naturalIdMapping == null ) {
-				// EARLY EXIT!!!
+			return;
-				// the natural id is mutable (!immutable), no need to do the below checks
-				return;
-			}
-
-			final int[] naturalIdentifierPropertiesIndexes = persister.getNaturalIdentifierProperties();
-			final Type[] propertyTypes = persister.getPropertyTypes();
-			final boolean[] propertyUpdateability = persister.getPropertyUpdateability();
-
-			final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
-			final Object[] snapshot = loaded == null
-					? persistenceContext.getNaturalIdSnapshot( entry.getId(), persister )
-					: persistenceContext.getNaturalIdHelper().extractNaturalIdValues( loaded, persister );
-
-			for ( int i = 0; i < naturalIdentifierPropertiesIndexes.length; i++ ) {
-				final int naturalIdentifierPropertyIndex = naturalIdentifierPropertiesIndexes[i];
-				if ( propertyUpdateability[naturalIdentifierPropertyIndex] ) {
-					// if the given natural id property is updatable (mutable), there is nothing to check
-					continue;
-				}
-
-				final Type propertyType = propertyTypes[naturalIdentifierPropertyIndex];
-				if ( !propertyType.isEqual( current[naturalIdentifierPropertyIndex], snapshot[i] ) ) {
-					throw new HibernateException(
-							String.format(
-									"An immutable natural identifier of entity %s was altered from `%s` to `%s`",
-									persister.getEntityName(),
-									propertyTypes[naturalIdentifierPropertyIndex].toLoggableString(
-											snapshot[i],
-											session.getFactory()
-									),
-									propertyTypes[naturalIdentifierPropertyIndex].toLoggableString(
-											current[naturalIdentifierPropertyIndex],
-											session.getFactory()
-									)
-							)
-					);
-				}
-			}
 		}
+
+		if ( entry.getStatus() == Status.READ_ONLY ) {
+			// nothing to check
+			return;
+		}
+
+		naturalIdMapping.verifyFlushState( entry.getId(), current, loaded, session );
 	}
 
 	/**

hibernate-core/src/main/java/org/hibernate/event/internal/DefaultLoadEventListener.java

@@ -538,7 +538,7 @@ private Object doLoad(
 		if ( entity != null && persister.hasNaturalIdentifier() ) {
 			final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
 			final PersistenceContext.NaturalIdHelper naturalIdHelper = persistenceContext.getNaturalIdHelper();
-			naturalIdHelper.cacheNaturalIdCrossReferenceFromLoad(
+			naturalIdHelper.cacheResolutionFromLoad(
 					persister,
 					event.getEntityId(),
 					naturalIdHelper.extractNaturalIdValues(

hibernate-core/src/main/java/org/hibernate/event/internal/DefaultResolveNaturalIdEventListener.java

@@ -136,7 +136,7 @@ protected Object loadFromDatasource(final ResolveNaturalIdEvent event) {
 		//PK can be null if the entity doesn't exist
 		if (pk != null) {
 			final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
-			persistenceContext.getNaturalIdHelper().cacheNaturalIdCrossReferenceFromLoad(
+			persistenceContext.getNaturalIdHelper().cacheResolutionFromLoad(
 					event.getEntityPersister(),
 					pk,
 					event.getOrderedNaturalIdValues()

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

@@ -75,7 +75,7 @@ public <K> List<E> multiLoad(K[] naturalIds, MultiNaturalIdLoadOptions options,
 				(naturalId, session1) -> {
 					// `naturalId` here is the one passed in by the API as part of the values array
 					// todo (6.0) : use this to help create the ordered results
-					return entityDescriptor.getNaturalIdMapping().normalizeValue( naturalId, session );
+					return entityDescriptor.getNaturalIdMapping().normalizeIncomingValue( naturalId, session );
 				},
 				session.getLoadQueryInfluencers(),
 				lockOptions,

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

@@ -47,7 +47,7 @@ interface KeyValueResolver {
 		 * the "true" form - single value for simple natural-ids and an array for
 		 * compound natural-ids.
 		 *
-		 * Generally delegates to {@link org.hibernate.metamodel.mapping.NaturalIdMapping#normalizeValue}
+		 * Generally delegates to {@link org.hibernate.metamodel.mapping.NaturalIdMapping#normalizeIncomingValue}
 		 */
 		Object resolveKeyToLoad(Object incoming, SharedSessionContractImplementor session);
 	}

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

@@ -61,7 +61,7 @@ protected void applyNaturalIdAsJdbcParameters(
 			SharedSessionContractImplementor session) {
 		assert jdbcParameters.size() == 1;
 
-		final Object bindableValue = naturalIdMapping().normalizeValue( naturalIdToLoad, session );
+		final Object bindableValue = naturalIdMapping().normalizeIncomingValue( naturalIdToLoad, session );
 
 		final SingularAttributeMapping attributeMapping = naturalIdMapping().getNaturalIdAttributes().get( 0 );
 		jdbcParamBindings.registerParametersForEachJdbcValue(
@@ -75,7 +75,7 @@ protected void applyNaturalIdAsJdbcParameters(
 
 	@Override
 	protected Object resolveNaturalIdBindValue(Object naturalIdToLoad, SharedSessionContractImplementor session) {
-		return naturalIdMapping().normalizeValue( naturalIdToLoad, session );
+		return naturalIdMapping().normalizeIncomingValue( naturalIdToLoad, session );
 	}
 
 	@Override
@@ -161,7 +161,7 @@ protected boolean isSimple() {
 	public Object resolveNaturalIdToId(
 			Object naturalIdValue,
 			SharedSessionContractImplementor session) {
-		final Object bindValue = naturalIdMapping().normalizeValue( naturalIdValue, session );
+		final Object bindValue = naturalIdMapping().normalizeIncomingValue( naturalIdValue, session );
 
 		final SessionFactoryImplementor sessionFactory = session.getFactory();
 		final JdbcServices jdbcServices = sessionFactory.getJdbcServices();

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

@@ -6,9 +6,11 @@
  */
 package org.hibernate.metamodel.mapping;
 
+import java.util.Objects;
 import java.util.function.BiConsumer;
 
 import org.hibernate.NotYetImplementedFor6Exception;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.metamodel.model.domain.NavigableRole;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.query.sqm.sql.internal.DomainResultProducer;
@@ -103,4 +105,9 @@ default int forEachSelection(int offset, SelectionConsumer consumer) {
 	}
 
 	EntityMappingType findContainingEntityMapping();
+
+	default boolean areEqual(Object one, Object other, SharedSessionContractImplementor session) {
+		// NOTE : deepEquals to account for arrays (compound natural-id)
+		return Objects.deepEquals( one, other );
+	}
 }

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

@@ -8,6 +8,7 @@
 
 import java.util.List;
 
+import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.loader.ast.spi.MultiNaturalIdLoader;
 import org.hibernate.loader.ast.spi.NaturalIdLoader;
@@ -23,6 +24,21 @@ public interface NaturalIdMapping extends VirtualModelPart {
 	 */
 	List<SingularAttributeMapping> getNaturalIdAttributes();
 
+	/**
+	 * Whether the natural-id is immutable.  This is the same as saying that none of
+	 * the attributes are mutable
+	 */
+	boolean isImmutable();
+
+	/**
+	 * Verify the natural-id value(s) we are about to flush to the database
+	 */
+	void verifyFlushState(
+			Object id,
+			Object[] currentState,
+			Object[] loadedState,
+			SharedSessionContractImplementor session);
+
 	@Override
 	default String getPartName() {
 		return PART_NAME;
@@ -31,5 +47,40 @@ default String getPartName() {
 	NaturalIdLoader getNaturalIdLoader();
 	MultiNaturalIdLoader getMultiNaturalIdLoader();
 
-	Object normalizeValue(Object incoming, SharedSessionContractImplementor session);
+	/**
+	 * Given an array of "full entity state", extract the normalized natural id representation
+	 *
+	 * @param state The attribute state array
+	 *
+	 * @return The extracted natural id values.  This is a normalized
+	 */
+	Object extractNaturalIdValues(Object[] state, SharedSessionContractImplementor session);
+
+	/**
+	 * Given an entity instance, extract the normalized natural id representation
+	 *
+	 * @param entity The entity instance
+	 *
+	 * @return The extracted natural id values
+	 */
+	Object extractNaturalIdValues(Object entity, SharedSessionContractImplementor session);
+
+	/**
+	 * Normalize an incoming (user supplied) natural-id value.
+	 */
+	Object normalizeIncomingValue(Object incoming, SharedSessionContractImplementor session);
+
+	/**
+	 * Validates a natural id value(s) for the described natural-id based on the expected internal representation
+	 */
+	void validateInternalForm(Object naturalIdValue, SharedSessionContractImplementor session);
+
+	/**
+	 * Calculate the hash-code of a natural-id value
+	 *
+	 * @param value The natural-id value
+	 *
+	 * @return The hash-code
+	 */
+	int calculateHashCode(Object value, SharedSessionContractImplementor session);
 }

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

@@ -31,6 +31,10 @@ public NavigableRole getNavigableRole() {
 		return role;
 	}
 
+	public EntityMappingType getDeclaringType() {
+		return declaringType;
+	}
+
 	@Override
 	public EntityMappingType findContainingEntityMapping() {
 		return declaringType;

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

@@ -7,11 +7,14 @@
 package org.hibernate.metamodel.mapping.internal;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.function.BiConsumer;
 
+import org.hibernate.HibernateException;
 import org.hibernate.NotYetImplementedFor6Exception;
+import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.loader.NaturalIdPostLoadListener;
 import org.hibernate.loader.NaturalIdPreLoadListener;
@@ -25,6 +28,7 @@
 import org.hibernate.metamodel.mapping.JdbcMapping;
 import org.hibernate.metamodel.mapping.MappingType;
 import org.hibernate.metamodel.mapping.SingularAttributeMapping;
+import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.ast.Clause;
 import org.hibernate.sql.ast.spi.SqlSelection;
@@ -42,6 +46,7 @@ public class CompoundNaturalIdMapping extends AbstractNaturalIdMapping implement
 
 	private final List<SingularAttributeMapping> attributes;
 	private final List<JdbcMapping> jdbcMappings;
+	private final boolean immutable;
 
 	private final NaturalIdLoader<?> loader;
 	private final MultiNaturalIdLoader<?> multiLoader;
@@ -54,11 +59,15 @@ public CompoundNaturalIdMapping(
 		super( declaringType, cacheRegionName );
 		this.attributes = attributes;
 
+		boolean anyMutable = false;
 		final List<JdbcMapping> jdbcMappings = new ArrayList<>();
 		for ( int i = 0; i < attributes.size(); i++ ) {
-			attributes.get( i ).forEachJdbcType( (index, jdbcMapping) -> jdbcMappings.add( jdbcMapping ) );
+			final SingularAttributeMapping attributeMapping = attributes.get( i );
+			attributeMapping.forEachJdbcType( (index, jdbcMapping) -> jdbcMappings.add( jdbcMapping ) );
+			anyMutable = anyMutable || attributeMapping.getAttributeMetadataAccess().resolveAttributeMetadata( null ).isUpdatable();
 		}
 		this.jdbcMappings = jdbcMappings;
+		this.immutable = ! anyMutable;
 
 		loader = new CompoundNaturalIdLoader<>(
 				this,
@@ -70,11 +79,43 @@ public CompoundNaturalIdMapping(
 		multiLoader = new MultiNaturalIdLoaderStandard<>( declaringType, creationProcess );
 	}
 
+	@Override
+	public Object[] extractNaturalIdValues(Object[] state, SharedSessionContractImplementor session) {
+		if ( state == null ) {
+			return null;
+		}
+
+		if ( state.length == attributes.size() ) {
+			return state;
+		}
+
+		final Object[] values = new Object[ attributes.size() ];
+
+		for ( int i = 0; i <= attributes.size() - 1; i++ ) {
+			final SingularAttributeMapping attributeMapping = attributes.get( i );
+			final Object domainValue = state[ attributeMapping.getStateArrayPosition() ];
+			values[ i ] = attributeMapping.disassemble( domainValue, session );
+		}
+
+		return values;
+	}
+
+	@Override
+	public Object[] extractNaturalIdValues(Object entity, SharedSessionContractImplementor session) {
+		final Object[] values = new Object[ attributes.size() ];
+
+		for ( int i = 0; i < attributes.size(); i++ ) {
+			values[i] = attributes.get( i ).getPropertyAccess().getGetter().get( entity );
+		}
+
+		return values;
+	}
+
 	@Override
 	@SuppressWarnings( "rawtypes" )
-	public Object normalizeValue(Object incoming, SharedSessionContractImplementor session) {
+	public Object[] normalizeIncomingValue(Object incoming, SharedSessionContractImplementor session) {
 		if ( incoming instanceof Object[] ) {
-			return incoming;
+			return (Object[]) incoming;
 		}
 
 		if ( incoming instanceof Map ) {
@@ -90,11 +131,95 @@ public Object normalizeValue(Object incoming, SharedSessionContractImplementor s
 		throw new UnsupportedOperationException( "Do not know how to normalize compound natural-id value : " + incoming );
 	}
 
+	@Override
+	public void validateInternalForm(Object naturalIdValue, SharedSessionContractImplementor session) {
+		if ( naturalIdValue == null ) {
+			return;
+		}
+
+		// should be an array, with a size equal to the number of attributes making up this compound natural-id
+		if ( naturalIdValue instanceof Object[] ) {
+			final Object[] values = (Object[]) naturalIdValue;
+			if ( values.length != attributes.size() ) {
+				throw new IllegalArgumentException(
+						"Natural-id value [" + naturalIdValue + "] did not contain the expected number of elements ["
+								+ attributes.size() + "]"
+				);
+			}
+
+			return;
+		}
+
+		throw new IllegalArgumentException( "Natural-id value [" + naturalIdValue + "] was not an array as expected" );
+	}
+
+	@Override
+	public int calculateHashCode(Object value, SharedSessionContractImplementor session) {
+		return 0;
+	}
+
+	@Override
+	public void verifyFlushState(Object id, Object[] currentState, Object[] loadedState, SharedSessionContractImplementor session) {
+		if ( ! immutable ) {
+			// EARLY EXIT!!!
+			// the natural id is mutable (!immutable), no need to do the checks
+			return;
+		}
+
+		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
+		final EntityPersister persister = getDeclaringType().getEntityPersister();
+
+		final Object[] naturalId = extractNaturalIdValues( currentState, session );
+
+		final Object snapshot = loadedState == null
+				? persistenceContext.getNaturalIdSnapshot( id, persister )
+				: persistenceContext.getNaturalIdHelper().extractNaturalIdValues( loadedState, persister );
+		final Object[] previousNaturalId = (Object[]) snapshot;
+
+		assert naturalId.length == getNaturalIdAttributes().size();
+		assert previousNaturalId.length == naturalId.length;
+
+		for ( int i = 0; i < getNaturalIdAttributes().size(); i++ ) {
+			final SingularAttributeMapping attributeMapping = getNaturalIdAttributes().get( i );
+
+			final boolean updatable = attributeMapping.getAttributeMetadataAccess().resolveAttributeMetadata( persister ).isUpdatable();
+			if ( updatable ) {
+				// property is updatable (mutable), there is nothing to check
+				continue;
+			}
+
+			final Object currentValue = naturalId[ i ];
+			final Object previousValue = previousNaturalId[ i ];
+
+			if ( ! attributeMapping.areEqual( currentValue, previousValue, session ) ) {
+				throw new HibernateException(
+						String.format(
+								"An immutable attribute [%s] within compound natural identifier of entity %s was altered from `%s` to `%s`",
+								attributeMapping.getAttributeName(),
+								persister.getEntityName(),
+								previousValue,
+								currentValue
+						)
+				);
+			}
+		}
+	}
+
+	@Override
+	public boolean areEqual(Object one, Object other, SharedSessionContractImplementor session) {
+		return Arrays.equals( (Object[]) one, (Object[]) other );
+	}
+
 	@Override
 	public List<SingularAttributeMapping> getNaturalIdAttributes() {
 		return attributes;
 	}
 
+	@Override
+	public boolean isImmutable() {
+		return immutable;
+	}
+
 	@Override
 	public NaturalIdLoader<?> getNaturalIdLoader() {
 		return loader;

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

@@ -11,6 +11,8 @@
 import java.util.Map;
 import java.util.function.BiConsumer;
 
+import org.hibernate.HibernateException;
+import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.loader.NaturalIdPostLoadListener;
 import org.hibernate.loader.NaturalIdPreLoadListener;
@@ -24,6 +26,7 @@
 import org.hibernate.metamodel.mapping.JdbcMapping;
 import org.hibernate.metamodel.mapping.MappingType;
 import org.hibernate.metamodel.mapping.SingularAttributeMapping;
+import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.query.NavigablePath;
 import org.hibernate.sql.ast.Clause;
 import org.hibernate.sql.ast.spi.SqlSelection;
@@ -38,6 +41,8 @@
 public class SimpleNaturalIdMapping extends AbstractNaturalIdMapping {
 	private final SingularAttributeMapping attribute;
 
+	private final boolean immutable;
+
 	private final SimpleNaturalIdLoader<?> loader;
 	private final MultiNaturalIdLoader<?> multiLoader;
 
@@ -49,6 +54,10 @@ public SimpleNaturalIdMapping(
 		super( declaringType, cacheRegionName );
 		this.attribute = attribute;
 
+		this.immutable = ! attribute.getAttributeMetadataAccess()
+				.resolveAttributeMetadata( null )
+				.isUpdatable();
+
 		this.loader = new SimpleNaturalIdLoader<>(
 				this,
 				NaturalIdPreLoadListener.NO_OP,
@@ -60,12 +69,77 @@ public SimpleNaturalIdMapping(
 	}
 
 	@Override
-	public Object normalizeValue(Object incoming, SharedSessionContractImplementor session) {
+	public void verifyFlushState(Object id, Object[] currentState, Object[] loadedState, SharedSessionContractImplementor session) {
-		return normalizeValue( incoming );
+		if ( ! immutable ) {
+			// EARLY EXIT!!!
+			// the natural id is mutable (!immutable), no need to do the checks
+			return;
+		}
+
+		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
+		final EntityPersister persister = getDeclaringType().getEntityPersister();
+
+		final Object naturalId = extractNaturalIdValues( currentState, session );
+		final Object snapshot = loadedState == null
+				? persistenceContext.getNaturalIdSnapshot( id, persister )
+				: persistenceContext.getNaturalIdHelper().extractNaturalIdValues( loadedState, persister );
+
+		if ( ! areEqual( naturalId, snapshot, session ) ) {
+			throw new HibernateException(
+					String.format(
+							"An immutable natural identifier of entity %s was altered from `%s` to `%s`",
+							persister.getEntityName(),
+							snapshot,
+							naturalId
+					)
+			);
+		}
+	}
+
+	@Override
+	public Object extractNaturalIdValues(Object[] state, SharedSessionContractImplementor session) {
+		return state[ attribute.getStateArrayPosition() ];
+	}
+
+	@Override
+	public Object extractNaturalIdValues(Object entity, SharedSessionContractImplementor session) {
+		return attribute.getPropertyAccess().getGetter().get( entity );
+	}
+
+	@Override
+	public void validateInternalForm(Object naturalIdValue, SharedSessionContractImplementor session) {
+		if ( naturalIdValue == null ) {
+			return;
+		}
+
+		if ( naturalIdValue.getClass().isArray() ) {
+			// be flexible
+			final Object[] values = (Object[]) naturalIdValue;
+			if ( values.length == 1 ) {
+				naturalIdValue = values[0];
+			}
+		}
+
+		if ( ! getJavaTypeDescriptor().getJavaType().isInstance( naturalIdValue ) ) {
+			throw new IllegalArgumentException(
+					"Incoming natural-id value [" + naturalIdValue + "] is not of expected type ["
+							+ getJavaTypeDescriptor().getJavaType().getName() + "]"
+			);
+		}
+	}
+
+	@Override
+	public int calculateHashCode(Object value, SharedSessionContractImplementor session) {
+		return 0;
+	}
+
+	@Override
+	public Object normalizeIncomingValue(Object incoming, SharedSessionContractImplementor session) {
+		return normalizeIncomingValue( incoming );
 	}
 
 	@SuppressWarnings( "rawtypes" )
-	public Object normalizeValue(Object naturalIdToLoad) {
+	public Object normalizeIncomingValue(Object naturalIdToLoad) {
 		if ( naturalIdToLoad instanceof Map ) {
 			final Map valueMap = (Map) naturalIdToLoad;
 			assert valueMap.size() == 1;
@@ -101,6 +175,11 @@ public List<SingularAttributeMapping> getNaturalIdAttributes() {
 		return Collections.singletonList( attribute );
 	}
 
+	@Override
+	public boolean isImmutable() {
+		return immutable;
+	}
+
 	@Override
 	public MappingType getPartMappingType() {
 		return attribute.getPartMappingType();

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

@@ -5010,7 +5010,7 @@ private void handleNaturalIdReattachment(Object entity, SharedSessionContractImp
 
 		// for reattachment of mutable natural-ids, we absolutely positively have to grab the snapshot from the
 		// database, because we have no other way to know if the state changed while detached.
-		final Object[] naturalIdSnapshot;
+		final Object naturalIdSnapshot;
 		final Object[] entitySnapshot = persistenceContext.getDatabaseSnapshot( id, this );
 		if ( entitySnapshot == StatefulPersistenceContext.NO_ROW ) {
 			naturalIdSnapshot = null;
@@ -5019,12 +5019,9 @@ private void handleNaturalIdReattachment(Object entity, SharedSessionContractImp
 			naturalIdSnapshot = naturalIdHelper.extractNaturalIdValues( entitySnapshot, this );
 		}
 
-		naturalIdHelper.removeSharedNaturalIdCrossReference( this, id, naturalIdSnapshot );
+		naturalIdHelper.removeSharedResolution( this, id, naturalIdSnapshot );
-		naturalIdHelper.manageLocalNaturalIdCrossReference(
+		naturalIdHelper.manageLocalResolution(
-				this,
+				id, naturalIdHelper.extractNaturalIdValues( entity, this ), this,
-				id,
-				naturalIdHelper.extractNaturalIdValues( entity, this ),
-				naturalIdSnapshot,
 				CachedNaturalIdValueSource.UPDATE
 		);
 	}

hibernate-core/src/main/java/org/hibernate/sql/results/graph/entity/AbstractEntityInitializer.java

@@ -659,7 +659,7 @@ public void initializeInstance(RowProcessingState rowProcessingState) {
 		}
 
 		if ( entityDescriptor.getNaturalIdMapping() != null ) {
-			persistenceContext.getNaturalIdHelper().cacheNaturalIdCrossReferenceFromLoad(
+			persistenceContext.getNaturalIdHelper().cacheResolutionFromLoad(
 					entityDescriptor,
 					entityIdentifier,
 					persistenceContext.getNaturalIdHelper()

hibernate-core/src/test/java/org/hibernate/cache/spi/NaturalIdCacheKeyTest.java

@@ -74,7 +74,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable {
         assertEquals(key.hashCode(), keyClone.hashCode());
         assertEquals(key.toString(), keyClone.toString());
         assertEquals(key.getEntityName(), keyClone.getEntityName());
-        assertArrayEquals(key.getNaturalIdValues(), keyClone.getNaturalIdValues());
+        assertEquals(key.getNaturalIdValues(), keyClone.getNaturalIdValues());
         assertEquals(key.getTenantId(), keyClone.getTenantId());
         
     }