@Override
@SuppressWarnings( {"unchecked"})
public EntityPersister createEntityPersister(
		PersistentClass entityBinding,
		EntityDataAccess entityCacheAccessStrategy,
		NaturalIdDataAccess naturalIdCacheAccessStrategy,
		PersisterCreationContext creationContext) throws HibernateException {
	// If the metadata for the entity specified an explicit persister class, use it...
	Class<? extends EntityPersister> persisterClass = entityBinding.getEntityPersisterClass();
	if ( persisterClass == null ) {
		// Otherwise, use the persister class indicated by the PersisterClassResolver service
		persisterClass = serviceRegistry.getService( PersisterClassResolver.class ).getEntityPersisterClass( entityBinding );
	}

	return createEntityPersister(
			persisterClass,
			entityBinding,
			entityCacheAccessStrategy,
			naturalIdCacheAccessStrategy,
			creationContext
	);
}
 

@Override
public NaturalIdDataAccess generateNaturalIdAccess(NaturalIdDataCachingConfig accessConfig) {
	final NavigableRole namedEntityRole = accessConfig.getNavigableRole();
	final AccessType accessType = accessConfig.getAccessType();

	log.debugf( "Generating entity natural-id access [%s] : %s", accessType.getExternalName(), namedEntityRole );

	switch ( accessType ) {
		case READ_ONLY: {
			return generateReadOnlyNaturalIdAccess( accessConfig );
		}
		case READ_WRITE: {
			return generateReadWriteNaturalIdAccess( accessConfig );
		}
		case NONSTRICT_READ_WRITE: {
			return generateNonStrictReadWriteNaturalIdAccess( accessConfig );
		}
		case TRANSACTIONAL: {
			return generateTransactionalNaturalIdDataAccess( accessConfig );
		}
		default: {
			throw new IllegalArgumentException( "Unrecognized cache AccessType - " + accessType );
		}
	}
}
 

public ReactiveSingleTableEntityPersister(
		PersistentClass persistentClass,
		EntityDataAccess cacheAccessStrategy,
		NaturalIdDataAccess naturalIdRegionAccessStrategy,
		PersisterCreationContext creationContext) throws HibernateException {
	super( persistentClass, cacheAccessStrategy, naturalIdRegionAccessStrategy, creationContext );
}
 

public ReactiveJoinedSubclassEntityPersister(
		PersistentClass persistentClass,
		EntityDataAccess cacheAccessStrategy,
		NaturalIdDataAccess naturalIdRegionAccessStrategy,
		PersisterCreationContext creationContext) throws HibernateException {
	super( persistentClass, cacheAccessStrategy, naturalIdRegionAccessStrategy, creationContext );
}
 

public ReactiveUnionSubclassEntityPersister(
		PersistentClass persistentClass,
		EntityDataAccess cacheAccessStrategy,
		NaturalIdDataAccess naturalIdRegionAccessStrategy,
		PersisterCreationContext creationContext) throws HibernateException {
	super( persistentClass, cacheAccessStrategy, naturalIdRegionAccessStrategy, creationContext );
}
 

/**
 * Handle removing cross reference entries for the given natural-id/pk combo
 *
 * @param persister The persister representing the entity type.
 * @param pk The primary key value
 * @param naturalIdValues The natural id value(s)
 * 
 * @return The cached values, if any.  May be different from incoming values.
 */
public Object[] removeNaturalIdCrossReference(EntityPersister persister, Serializable pk, Object[] naturalIdValues) {
	persister = locatePersisterForKey( persister );
	validateNaturalId( persister, naturalIdValues );

	final NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );
	Object[] sessionCachedNaturalIdValues = null;
	if ( entityNaturalIdResolutionCache != null ) {
		final CachedNaturalId cachedNaturalId = entityNaturalIdResolutionCache.pkToNaturalIdMap
				.remove( pk );
		if ( cachedNaturalId != null ) {
			entityNaturalIdResolutionCache.naturalIdToPkMap.remove( cachedNaturalId );
			sessionCachedNaturalIdValues = cachedNaturalId.getValues();
		}
	}

	if ( persister.hasNaturalIdCache() ) {
		final NaturalIdDataAccess naturalIdCacheAccessStrategy = persister
				.getNaturalIdCacheAccessStrategy();
		final Object naturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( naturalIdValues, persister, session() );
		naturalIdCacheAccessStrategy.evict( naturalIdCacheKey );

		if ( sessionCachedNaturalIdValues != null
				&& !Arrays.equals( sessionCachedNaturalIdValues, naturalIdValues ) ) {
			final Object sessionNaturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( sessionCachedNaturalIdValues, persister, session() );
			naturalIdCacheAccessStrategy.evict( sessionNaturalIdCacheKey );
		}
	}

	return sessionCachedNaturalIdValues;
}
 

@Override
		public void removeSharedNaturalIdCrossReference(EntityPersister persister, Serializable id, Object[] naturalIdValues) {
			if ( !persister.hasNaturalIdentifier() ) {
				// nothing to do
				return;
			}

			if ( ! persister.hasNaturalIdCache() ) {
				// nothing to do
				return;
			}

			// todo : couple of things wrong here:
			//		1) should be using access strategy, not plain evict..
			//		2) should prefer session-cached values if any (requires interaction from removeLocalNaturalIdCrossReference

			persister = locateProperPersister( persister );
			final NaturalIdDataAccess naturalIdCacheAccessStrategy = persister.getNaturalIdCacheAccessStrategy();
			final Object naturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( naturalIdValues, persister, session );
			naturalIdCacheAccessStrategy.evict( naturalIdCacheKey );

//			if ( sessionCachedNaturalIdValues != null
//					&& !Arrays.equals( sessionCachedNaturalIdValues, deletedNaturalIdValues ) ) {
//				final NaturalIdCacheKey sessionNaturalIdCacheKey = new NaturalIdCacheKey( sessionCachedNaturalIdValues, persister, session );
//				naturalIdCacheAccessStrategy.evict( sessionNaturalIdCacheKey );
//			}
		}
 

DeprecatedNaturalIdCacheStatisticsImpl(String regionName, Set<NaturalIdDataAccess> accessStrategies) {
	this.regionName = regionName;
	this.accessStrategies = accessStrategies;
	final ReadWriteLock lock = new ReentrantReadWriteLock();
	this.readLock = lock.readLock();
	this.writeLock = lock.writeLock();
}
 

@Override
public long getElementCountInMemory() {
	long count = 0;
	HashSet<Region> processedRegions = null;

	for ( NaturalIdDataAccess accessStrategy : accessStrategies ) {
		final DomainDataRegion region = accessStrategy.getRegion();
		if ( ExtendedStatisticsSupport.class.isInstance( region ) ) {

		}

		if ( region instanceof ExtendedStatisticsSupport ) {
			if ( processedRegions == null ) {
				processedRegions = new HashSet<>();
			}
			if ( processedRegions.add( region ) ) {
				count += ( (ExtendedStatisticsSupport) region ).getElementCountInMemory();
			}
		}

	}

	if ( count == 0 ) {
		return NO_EXTENDED_STAT_SUPPORT_RETURN;
	}

	return count;
}
 

@Override
public long getElementCountOnDisk() {
	long count = 0;
	HashSet<Region> processedRegions = null;

	for ( NaturalIdDataAccess accessStrategy : accessStrategies ) {
		final DomainDataRegion region = accessStrategy.getRegion();
		if ( ExtendedStatisticsSupport.class.isInstance( region ) ) {

		}

		if ( region instanceof ExtendedStatisticsSupport ) {
			if ( processedRegions == null ) {
				processedRegions = new HashSet<>();
			}
			if ( processedRegions.add( region ) ) {
				count += ( (ExtendedStatisticsSupport) region ).getElementCountOnDisk();
			}
		}

	}

	if ( count == 0 ) {
		return NO_EXTENDED_STAT_SUPPORT_RETURN;
	}

	return count;
}
 

@Override
public long getSizeInMemory() {
	long count = 0;
	HashSet<Region> processedRegions = null;

	for ( NaturalIdDataAccess accessStrategy : accessStrategies ) {
		final DomainDataRegion region = accessStrategy.getRegion();
		if ( ExtendedStatisticsSupport.class.isInstance( region ) ) {

		}

		if ( region instanceof ExtendedStatisticsSupport ) {
			if ( processedRegions == null ) {
				processedRegions = new HashSet<>();
			}
			if ( processedRegions.add( region ) ) {
				count += ( (ExtendedStatisticsSupport) region ).getElementCountOnDisk();
			}
		}

	}

	if ( count == 0 ) {
		return NO_EXTENDED_STAT_SUPPORT_RETURN;
	}

	return count;
}
 

@SuppressWarnings("WeakerAccess")
protected NaturalIdDataAccess generateReadOnlyNaturalIdAccess(NaturalIdDataCachingConfig accessConfig) {
	return new NaturalIdReadOnlyAccess(
			this,
			getEffectiveKeysFactory(),
			getCacheStorageAccess(),
			accessConfig
	);
}
 

@SuppressWarnings("WeakerAccess")
protected NaturalIdDataAccess generateReadWriteNaturalIdAccess(NaturalIdDataCachingConfig accessConfig) {
	return new NaturalIdReadWriteAccess(
			this,
			getEffectiveKeysFactory(),
			getCacheStorageAccess(),
			accessConfig
	);
}
 

@SuppressWarnings("WeakerAccess")
protected NaturalIdDataAccess generateNonStrictReadWriteNaturalIdAccess(NaturalIdDataCachingConfig accessConfig) {
	return new NaturalIdNonStrictReadWriteAccess(
			this,
			getEffectiveKeysFactory(),
			getCacheStorageAccess(),
			accessConfig
	);
}
 

@Override
public NaturalIdDataAccess getNaturalIdDataAccess(NavigableRole rootEntityRole) {
	final NaturalIdDataAccess access = naturalIdDataAccessMap.get( rootEntityRole );
	if ( access == null ) {
		throw new IllegalArgumentException( "Caching was not configured for entity natural-id : " + rootEntityRole.getFullPath() );
	}
	return access;
}
 

private Map<NavigableRole, NaturalIdDataAccess> generateNaturalIdDataAccessMap(DomainDataRegionConfig regionConfig) {
	if ( regionConfig.getNaturalIdCaching().isEmpty() ) {
		return Collections.emptyMap();
	}

	final Map<NavigableRole, NaturalIdDataAccess> accessMap = new ConcurrentHashMap<>();
	for ( NaturalIdDataCachingConfig naturalIdAccessConfig : regionConfig.getNaturalIdCaching() ) {
		accessMap.computeIfAbsent(
				naturalIdAccessConfig.getNavigableRole(),
				hierarchy -> generateNaturalIdAccess( naturalIdAccessConfig )
		);
	}

	return Collections.unmodifiableMap( accessMap );
}
 

@Override
protected NaturalIdDataAccess generateTransactionalNaturalIdDataAccess(NaturalIdDataCachingConfig accessConfig) {
	return new NaturalIdTransactionalAccess(
			this,
			getEffectiveKeysFactory(),
			getCacheStorageAccess(),
			accessConfig
	);
}
 

private void evictNaturalIdData(NavigableRole rootEntityRole, NaturalIdDataAccess cacheAccess) {
	if ( cacheAccess == null ) {
		return;
	}

	if ( LOG.isDebugEnabled() ) {
		LOG.debugf( "Evicting natural-id cache: %s", rootEntityRole.getFullPath() );
	}

	cacheAccess.evictAll();
}
 

public NaturalIdCleanup(
		NaturalIdDataAccess naturalIdCacheAccessStrategy,
		SharedSessionContractImplementor session) {
	this.naturalIdCacheAccessStrategy = naturalIdCacheAccessStrategy;
	this.cacheLock = naturalIdCacheAccessStrategy.lockRegion();
	naturalIdCacheAccessStrategy.removeAll( session );
}
 

/** {@inheritDoc} */
@Override protected NaturalIdDataAccess generateNaturalIdAccess(NaturalIdDataCachingConfig naturalIdAccessCfg) {
    HibernateCacheProxy cache = stgyFactory.regionCache(getName());
    AccessType accessType = naturalIdAccessCfg.getAccessType();
    Ignite ignite = stgyFactory.node();
    switch (accessType) {
        case READ_ONLY:
            HibernateAccessStrategyAdapter readOnlyStgy =
                stgyFactory.createReadOnlyStrategy(cache);
            return new IgniteNaturalIdDataAccess(readOnlyStgy, accessType, getRegionFactory(),
                this, ignite, cache);

        case NONSTRICT_READ_WRITE:
            HibernateAccessStrategyAdapter nonStrictReadWriteStgy =
                stgyFactory.createNonStrictReadWriteStrategy(cache);
            return new IgniteNaturalIdDataAccess(nonStrictReadWriteStgy, accessType, getRegionFactory(),
                this, ignite, cache);

        case READ_WRITE:
            HibernateAccessStrategyAdapter readWriteStgy =
                stgyFactory.createReadWriteStrategy(cache);
            return new IgniteNaturalIdDataAccess(readWriteStgy, accessType, getRegionFactory(),
                this, ignite, cache);

        case TRANSACTIONAL:
            HibernateAccessStrategyAdapter transactionalStgy =
                stgyFactory.createTransactionalStrategy(cache);
            return new IgniteNaturalIdDataAccess(transactionalStgy, accessType, getRegionFactory(),
                this, ignite, cache);

        default:
            throw new IllegalArgumentException("Unknown Hibernate access type: " + accessType);
    }
}
 

public NaturalIdDataAccess getNaturalIdCacheAccessStrategy() {
	return naturalIdRegionAccessStrategy;
}
 

/**
 * Given a persister and natural-id value(s), find the locally cross-referenced primary key.  Will return
 * {@link PersistenceContext.NaturalIdHelper#INVALID_NATURAL_ID_REFERENCE} if the given natural ids are known to
 * be invalid (see {@link #stashInvalidNaturalIdReference}).
 *
 * @param persister The persister representing the entity type.
 * @param naturalIdValues The natural id value(s)
 * 
 * @return The corresponding cross-referenced primary key, 
 * 		{@link PersistenceContext.NaturalIdHelper#INVALID_NATURAL_ID_REFERENCE},
 * 		or {@code null} if none 
 */
public Serializable findCachedNaturalIdResolution(EntityPersister persister, Object[] naturalIdValues) {
	persister = locatePersisterForKey( persister );
	validateNaturalId( persister, naturalIdValues );

	NaturalIdResolutionCache entityNaturalIdResolutionCache = naturalIdResolutionCacheMap.get( persister );

	Serializable pk;
	final CachedNaturalId cachedNaturalId = new CachedNaturalId( persister, naturalIdValues );
	if ( entityNaturalIdResolutionCache != null ) {
		pk = entityNaturalIdResolutionCache.naturalIdToPkMap.get( cachedNaturalId );

		// Found in session cache
		if ( pk != null ) {
			if ( LOG.isTraceEnabled() ) {
				LOG.trace(
						"Resolved natural key -> primary key resolution in session cache: " +
								persister.getRootEntityName() + "#[" +
								Arrays.toString( naturalIdValues ) + "]"
				);
			}

			return pk;
		}

		// if we did not find a hit, see if we know about these natural ids as invalid...
		if ( entityNaturalIdResolutionCache.containsInvalidNaturalIdReference( naturalIdValues ) ) {
			return PersistenceContext.NaturalIdHelper.INVALID_NATURAL_ID_REFERENCE;
		}
	}

	// Session cache miss, see if second-level caching is enabled
	if ( !persister.hasNaturalIdCache() ) {
		return null;
	}

	// Try resolution from second-level cache
	final NaturalIdDataAccess naturalIdCacheAccessStrategy = persister.getNaturalIdCacheAccessStrategy();
	final Object naturalIdCacheKey = naturalIdCacheAccessStrategy.generateCacheKey( naturalIdValues, persister, session() );

	pk = CacheHelper.fromSharedCache( session(), naturalIdCacheKey, naturalIdCacheAccessStrategy );

	// Found in second-level cache, store in session cache
	final SessionFactoryImplementor factory = session().getFactory();
	if ( pk != null ) {
		if ( factory.getStatistics().isStatisticsEnabled() ) {
			factory.getStatistics().naturalIdCacheHit(
					StatsHelper.INSTANCE.getRootEntityRole( persister ),
					naturalIdCacheAccessStrategy.getRegion().getName()
			);
		}

		if ( LOG.isTraceEnabled() ) {
			// 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 ),
					pk,
					persister.getRootEntityName()
			);
		}

		if ( entityNaturalIdResolutionCache == null ) {
			entityNaturalIdResolutionCache = new NaturalIdResolutionCache( persister );
			NaturalIdResolutionCache existingCache = naturalIdResolutionCacheMap.putIfAbsent( persister, entityNaturalIdResolutionCache );
			if ( existingCache != null ) {
				entityNaturalIdResolutionCache = existingCache;
			}
		}

		entityNaturalIdResolutionCache.pkToNaturalIdMap.put( pk, cachedNaturalId );
		entityNaturalIdResolutionCache.naturalIdToPkMap.put( cachedNaturalId, pk );
	}
	else if ( factory.getStatistics().isStatisticsEnabled() ) {
		factory.getStatistics().naturalIdCacheMiss(
				StatsHelper.INSTANCE.getRootEntityRole( persister ),
				naturalIdCacheAccessStrategy.getRegion().getName()
		);
	}

	return pk;
}
 

@SuppressWarnings({"WeakerAccess"})
protected NaturalIdDataAccess generateTransactionalNaturalIdDataAccess(NaturalIdDataCachingConfig accessConfig) {
	throw generateTransactionalNotSupportedException();
}
 

protected void releaseDataAccess(NaturalIdDataAccess cacheAccess) {
	if ( Destructible.class.isInstance( cacheAccess ) ) {
		( (Destructible) cacheAccess ).destroy();
	}
}
 

@Override
public NaturalIdDataAccess getNaturalIdCacheRegionAccessStrategy(NavigableRole rootEntityName) {
	return null;
}
 

@Override
public Set<NaturalIdDataAccess> getNaturalIdAccessesInRegion(String regionName) {
	return Collections.emptySet();
}
 

@Override
public void prime(Set<DomainDataRegionConfig> cacheRegionConfigs) {
	for ( DomainDataRegionConfig regionConfig : cacheRegionConfigs ) {
		final DomainDataRegion region = getRegionFactory().buildDomainDataRegion( regionConfig, this );
		regionsByName.put( region.getName(), region );

		if ( ! Objects.equals( region.getName(), regionConfig.getRegionName() ) ) {
			throw new HibernateException(
					String.format(
							Locale.ROOT,
							"Region [%s] returned from RegionFactory [%s] was named differently than requested name.  Expecting `%s`, but found `%s`",
							region,
							getRegionFactory().getClass().getName(),
							regionConfig.getRegionName(),
							region.getName()
					)
			);
		}

		// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
		// Entity caching

		for ( EntityDataCachingConfig entityAccessConfig : regionConfig.getEntityCaching() ) {
			final EntityDataAccess entityDataAccess = entityAccessMap.put(
					entityAccessConfig.getNavigableRole(),
					region.getEntityDataAccess( entityAccessConfig.getNavigableRole() )
			);

			legacySecondLevelCacheNames.add(
					StringHelper.qualifyConditionally(
							getSessionFactory().getSessionFactoryOptions().getCacheRegionPrefix(),
							region.getName()
					)
			);
		}


		// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
		// Natural-id caching

		if ( regionConfig.getNaturalIdCaching().isEmpty() ) {
			legacyNaturalIdAccessesForRegion.put( region.getName(), Collections.emptySet() );
		}
		else {
			final HashSet<NaturalIdDataAccess> accesses = new HashSet<>();

			for ( NaturalIdDataCachingConfig naturalIdAccessConfig : regionConfig.getNaturalIdCaching() ) {
				final NaturalIdDataAccess naturalIdDataAccess = naturalIdAccessMap.put(
						naturalIdAccessConfig.getNavigableRole(),
						region.getNaturalIdDataAccess( naturalIdAccessConfig.getNavigableRole() )
				);
				accesses.add( naturalIdDataAccess );
			}

			legacyNaturalIdAccessesForRegion.put( region.getName(), accesses );
		}


		// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
		// Collection caching

		for ( CollectionDataCachingConfig collectionAccessConfig : regionConfig.getCollectionCaching() ) {
			final CollectionDataAccess collectionDataAccess = collectionAccessMap.put(
					collectionAccessConfig.getNavigableRole(),
					region.getCollectionDataAccess( collectionAccessConfig.getNavigableRole() )
			);

			legacySecondLevelCacheNames.add(
					StringHelper.qualifyConditionally(
							getSessionFactory().getSessionFactoryOptions().getCacheRegionPrefix(),
							region.getName()
					)
			);
		}
	}

}
 

@Override
public NaturalIdDataAccess getNaturalIdCacheRegionAccessStrategy(NavigableRole rootEntityName) {
	return naturalIdAccessMap.get( rootEntityName );
}
 

@Override
public Set<NaturalIdDataAccess> getNaturalIdAccessesInRegion(String regionName) {
	return legacyNaturalIdAccessesForRegion.get( regionName );
}
 

/**
 * Get the NaturalId cache (optional operation)
 */
NaturalIdDataAccess getNaturalIdCacheAccessStrategy();