HHH-15509 deletion of unloaded entity

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

@@ -11,6 +11,7 @@ import org.hibernate.HibernateException;
 import org.hibernate.cache.spi.access.EntityDataAccess;
 import org.hibernate.cache.spi.access.SoftLock;
 import org.hibernate.engine.spi.EntityEntry;
+import org.hibernate.engine.spi.EntityKey;
 import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SessionImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
@@ -105,7 +106,9 @@ public class EntityDeleteAction extends EntityAction {
 		final boolean veto = preDelete();
 
 		Object version = this.version;
-		if ( persister.isVersionPropertyGenerated() ) {
+		if ( persister.isVersionPropertyGenerated()
+				// null instance signals that we're deleting an unloaded proxy, no need for a version
+				&& instance != null ) {
 			// we need to grab the version value from the entity, otherwise
 			// we have issues with generated-version entities that may have
 			// multiple actions queued during the same flush
@@ -125,9 +128,28 @@ public class EntityDeleteAction extends EntityAction {
 		if ( !isCascadeDeleteEnabled && !veto ) {
 			persister.delete( id, version, instance, session );
 		}
-		
+
-		//postDelete:
+		if ( instance == null ) {
-		// After actually deleting a row, record the fact that the instance no longer 
+			// null instance signals that we're deleting an unloaded proxy
+			postDeleteUnloaded( id, persister, session, ck );
+		}
+		else {
+			postDeleteLoaded( id, persister, session, instance, ck );
+		}
+
+		final StatisticsImplementor statistics = getSession().getFactory().getStatistics();
+		if ( statistics.isStatisticsEnabled() && !veto ) {
+			statistics.deleteEntity( getPersister().getEntityName() );
+		}
+	}
+
+	private void postDeleteLoaded(
+			Object id,
+			EntityPersister persister,
+			SharedSessionContractImplementor session,
+			Object instance,
+			Object ck) {
+		// After actually deleting a row, record the fact that the instance no longer
 		// exists on the database (needed for identity-column key generation), and
 		// remove it from the session cache
 		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
@@ -137,20 +159,28 @@ public class EntityDeleteAction extends EntityAction {
 		}
 		entry.postDelete();
 
-		persistenceContext.removeEntity( entry.getEntityKey() );
+		EntityKey key = entry.getEntityKey();
-		persistenceContext.removeProxy( entry.getEntityKey() );
+		persistenceContext.removeEntity( key );
-		
+		persistenceContext.removeProxy( key );
+
 		if ( persister.canWriteToCache() ) {
-			persister.getCacheAccessStrategy().remove( session, ck);
+			persister.getCacheAccessStrategy().remove( session, ck );
 		}
 
 		persistenceContext.getNaturalIdResolutions().removeSharedResolution( id, naturalIdValues, persister );
 
 		postDelete();
+	}
 
-		final StatisticsImplementor statistics = getSession().getFactory().getStatistics();
+	private void postDeleteUnloaded(Object id, EntityPersister persister, SharedSessionContractImplementor session, Object ck) {
-		if ( statistics.isStatisticsEnabled() && !veto ) {
+		final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
-			statistics.deleteEntity( getPersister().getEntityName() );
+		EntityKey key = session.generateEntityKey( id, persister );
+		if ( !persistenceContext.containsDeletedUnloadedEntityKey( key ) ) {
+			throw new AssertionFailure( "deleted proxy should be for an unloaded entity: " + key );
+		}
+		persistenceContext.removeProxy( key );
+		if ( persister.canWriteToCache() ) {
+			persister.getCacheAccessStrategy().remove( session, ck );
 		}
 	}
 

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

@@ -10,6 +10,7 @@ import org.hibernate.HibernateException;
 import org.hibernate.TransientObjectException;
 import org.hibernate.bytecode.enhance.spi.LazyPropertyInitializer;
 import org.hibernate.engine.spi.EntityEntry;
+import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SelfDirtinessTracker;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.internal.util.StringHelper;
@@ -195,17 +196,28 @@ public final class ForeignKeys {
 				return false;
 			}
 
+			final PersistenceContext persistenceContext = session.getPersistenceContextInternal();
+
 			if ( object instanceof HibernateProxy ) {
-				// if its an uninitialized proxy it can't be transient
+				// if it's an uninitialized proxy it can only be
+				// transient if we did an unloaded-delete on the
+				// proxy itself, in which case there is no entry
+				// for it, but its key has already been registered
+				// as nullifiable
 				final LazyInitializer li = ( (HibernateProxy) object ).getHibernateLazyInitializer();
-				if ( li.getImplementation( session ) == null ) {
+				Object entity = li.getImplementation( session );
-					return false;
+				if ( entity == null ) {
-					// ie. we never have to null out a reference to
+					return persistenceContext.containsDeletedUnloadedEntityKey(
-					// an uninitialized proxy
+							session.generateEntityKey(
+									li.getIdentifier(),
+									session.getFactory().getRuntimeMetamodels().getMappingMetamodel()
+											.getEntityDescriptor( li.getEntityName() )
+							)
+					);
 				}
 				else {
 					//unwrap it
-					object = li.getImplementation( session );
+					object = entity;
 				}
 			}
 
@@ -214,7 +226,7 @@ public final class ForeignKeys {
 			// case we definitely need to nullify
 			if ( object == self ) {
 				return isEarlyInsert
-					|| ( isDelete && session.getFactory().getJdbcServices().getDialect().hasSelfReferentialForeignKeyBug() );
+					|| isDelete && session.getFactory().getJdbcServices().getDialect().hasSelfReferentialForeignKeyBug();
 			}
 
 			// See if the entity is already bound to this session, if not look at the
@@ -222,13 +234,10 @@ public final class ForeignKeys {
 			// id is not "unsaved" (that is, we rely on foreign keys to keep
 			// database integrity)
 
-			final EntityEntry entityEntry = session.getPersistenceContextInternal().getEntry( object );
+			final EntityEntry entityEntry = persistenceContext.getEntry( object );
-			if ( entityEntry == null ) {
+			return entityEntry == null
-				return isTransient( entityName, object, null, session );
+					? isTransient( entityName, object, null, session )
-			}
+					: entityEntry.isNullifiable( isEarlyInsert, session );
-			else {
-				return entityEntry.isNullifiable( isEarlyInsert, session );
-			}
 		}
 	}
 
@@ -245,19 +254,11 @@ public final class ForeignKeys {
 	 *
 	 * @return {@code true} if the given entity is not transient (meaning it is either detached/persistent)
 	 */
-	@SuppressWarnings("SimplifiableIfStatement")
 	public static boolean isNotTransient(String entityName, Object entity, Boolean assumed, SharedSessionContractImplementor session) {
-		if ( entity instanceof HibernateProxy ) {
+		return entity instanceof HibernateProxy
-			return true;
+			|| session.getPersistenceContextInternal().isEntryFor( entity )
-		}
+			// todo : shouldn't assumed be reversed here?
-
+			|| !isTransient( entityName, entity, assumed, session );
-		if ( session.getPersistenceContextInternal().isEntryFor( entity ) ) {
-			return true;
-		}
-
-		// todo : shouldn't assumed be reversed here?
-
-		return !isTransient( entityName, entity, assumed, session );
 	}
 
 	/**
@@ -305,7 +306,6 @@ public final class ForeignKeys {
 				persister
 		);
 		return snapshot == null;
-
 	}
 
 	/**

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

@@ -127,6 +127,9 @@ public class StatefulPersistenceContext implements PersistenceContext {
 	// Set of EntityKeys of deleted objects
 	private HashSet<EntityKey> nullifiableEntityKeys;
 
+	// Set of EntityKeys of deleted unloaded proxies
+	private HashSet<EntityKey> deletedUnloadedEntityKeys;
+
 	// properties that we have tried to load, and not found in the database
 	private HashSet<AssociationKey> nullAssociations;
 
@@ -252,6 +255,7 @@ public class StatefulPersistenceContext implements PersistenceContext {
 		unownedCollections = null;
 		proxiesByKey = null;
 		nullifiableEntityKeys = null;
+		deletedUnloadedEntityKeys = null;
 		if ( batchFetchQueue != null ) {
 			batchFetchQueue.clear();
 		}
@@ -1600,6 +1604,7 @@ public class StatefulPersistenceContext implements PersistenceContext {
 				}
 		);
 		writeCollectionToStream( nullifiableEntityKeys, oos, "nullifiableEntityKey", EntityKey::serialize );
+		writeCollectionToStream( deletedUnloadedEntityKeys, oos, "deletedUnloadedEntityKeys", EntityKey::serialize );
 	}
 
 	private interface Serializer<E> {
@@ -1759,6 +1764,15 @@ public class StatefulPersistenceContext implements PersistenceContext {
 			for ( int i = 0; i < count; i++ ) {
 				rtn.nullifiableEntityKeys.add( EntityKey.deserialize( ois, sfi ) );
 			}
+			count = ois.readInt();
+
+			if ( LOG.isTraceEnabled() ) {
+				LOG.trace( "Starting deserialization of [" + count + "] deletedUnloadedEntityKeys entries" );
+			}
+			rtn.deletedUnloadedEntityKeys = new HashSet<>();
+			for ( int i = 0; i < count; i++ ) {
+				rtn.deletedUnloadedEntityKeys.add( EntityKey.deserialize( ois, sfi ) );
+			}
 
 		}
 		catch ( HibernateException he ) {
@@ -1829,7 +1843,7 @@ public class StatefulPersistenceContext implements PersistenceContext {
 		if ( nullifiableEntityKeys == null ) {
 			nullifiableEntityKeys = new HashSet<>();
 		}
-		this.nullifiableEntityKeys.add( key );
+		nullifiableEntityKeys.add( key );
 	}
 
 	@Override
@@ -1838,6 +1852,20 @@ public class StatefulPersistenceContext implements PersistenceContext {
 			|| nullifiableEntityKeys.size() == 0;
 	}
 
+	@Override
+	public boolean containsDeletedUnloadedEntityKey(EntityKey ek) {
+		return deletedUnloadedEntityKeys != null
+			&& deletedUnloadedEntityKeys.contains( ek );
+	}
+
+	@Override
+	public void registerDeletedUnloadedEntityKey(EntityKey key) {
+		if ( deletedUnloadedEntityKeys == null ) {
+			deletedUnloadedEntityKeys = new HashSet<>();
+		}
+		deletedUnloadedEntityKeys.add( key );
+	}
+
 	@Override
 	public int getCollectionEntriesSize() {
 		return collectionEntries == null ? 0 : collectionEntries.size();
@@ -1851,9 +1879,10 @@ public class StatefulPersistenceContext implements PersistenceContext {
 	@Override
 	public void clearCollectionsByKey() {
 		if ( collectionsByKey != null ) {
-			//A valid alternative would be to set this to null, like we do on close.
+			// A valid alternative would be to set this to null, like we do on close.
-			//The difference being that in this case we expect the collection will be used again, so we bet that clear()
+			// The difference being that in this case we expect the collection will be
-			//might allow us to skip having to re-allocate the collection.
+			// used again, so we bet that clear() might allow us to skip having to
+			// re-allocate the collection.
 			collectionsByKey.clear();
 		}
 	}
@@ -1863,8 +1892,7 @@ public class StatefulPersistenceContext implements PersistenceContext {
 		if ( collectionsByKey == null ) {
 			collectionsByKey = CollectionHelper.mapOfSize( INIT_COLL_SIZE );
 		}
-		final PersistentCollection<?> old = collectionsByKey.put( collectionKey, persistentCollection );
+		return collectionsByKey.put( collectionKey, persistentCollection );
-		return old;
 	}
 
 	@Override
@@ -1888,7 +1916,7 @@ public class StatefulPersistenceContext implements PersistenceContext {
 	@Override
 	public NaturalIdResolutions getNaturalIdResolutions() {
 		if ( naturalIdResolutions == null ) {
-			this.naturalIdResolutions = new NaturalIdResolutionsImpl( this );
+			naturalIdResolutions = new NaturalIdResolutionsImpl( this );
 		}
 		return naturalIdResolutions;
 	}

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

@@ -748,6 +748,10 @@ public interface PersistenceContext {
 	 */
 	boolean isNullifiableEntityKeysEmpty();
 
+	boolean containsDeletedUnloadedEntityKey(EntityKey ek);
+
+	void registerDeletedUnloadedEntityKey(EntityKey key);
+
 	/**
 	 * The size of the internal map storing all collection entries.
 	 * (The map is not exposed directly, but the size is often useful)

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

@@ -19,12 +19,10 @@ import org.hibernate.query.Query;
 import org.hibernate.SharedSessionContract;
 import org.hibernate.Transaction;
 import org.hibernate.cache.spi.CacheTransactionSynchronization;
-import org.hibernate.cfg.Environment;
 import org.hibernate.collection.spi.PersistentCollection;
 import org.hibernate.engine.jdbc.LobCreationContext;
 import org.hibernate.engine.jdbc.spi.JdbcCoordinator;
 import org.hibernate.engine.jdbc.spi.JdbcServices;
-import org.hibernate.internal.util.config.ConfigurationHelper;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.query.spi.QueryProducerImplementor;
 import org.hibernate.resource.jdbc.spi.JdbcSessionOwner;
@@ -384,7 +382,7 @@ public interface SharedSessionContractImplementor
 	 *
 	 * If the Session-level JDBC batch size was not configured, return the SessionFactory-level one.
 	 *
-	 * @return Session-level or or SessionFactory-level JDBC batch size.
+	 * @return Session-level or SessionFactory-level JDBC batch size.
 	 *
 	 * @since 5.2
 	 *
@@ -393,14 +391,9 @@ public interface SharedSessionContractImplementor
 	 */
 	default Integer getConfiguredJdbcBatchSize() {
 		final Integer sessionJdbcBatchSize = getJdbcBatchSize();
-
+		return sessionJdbcBatchSize == null
-		return sessionJdbcBatchSize == null ?
+				? getFactory().getSessionFactoryOptions().getJdbcBatchSize()
-			ConfigurationHelper.getInt(
+				: sessionJdbcBatchSize;
-					Environment.STATEMENT_BATCH_SIZE,
-					getFactory().getProperties(),
-					1
-			) :
-			sessionJdbcBatchSize;
 	}
 
 	/**

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

@@ -7,6 +7,7 @@
 package org.hibernate.event.internal;
 
 import org.hibernate.CacheMode;
+import org.hibernate.EmptyInterceptor;
 import org.hibernate.HibernateException;
 import org.hibernate.LockMode;
 import org.hibernate.TransientObjectException;
@@ -32,12 +33,16 @@ import org.hibernate.event.spi.DeleteEventListener;
 import org.hibernate.event.spi.EventSource;
 import org.hibernate.internal.CoreLogging;
 import org.hibernate.internal.CoreMessageLogger;
+import org.hibernate.internal.FastSessionServices;
 import org.hibernate.jpa.event.spi.CallbackRegistry;
 import org.hibernate.jpa.event.spi.CallbackRegistryConsumer;
+import org.hibernate.jpa.event.spi.CallbackType;
 import org.hibernate.persister.collection.CollectionPersister;
 import org.hibernate.persister.entity.EntityPersister;
 import org.hibernate.pretty.MessageHelper;
 import org.hibernate.property.access.internal.PropertyAccessStrategyBackRefImpl;
+import org.hibernate.proxy.HibernateProxy;
+import org.hibernate.proxy.LazyInitializer;
 import org.hibernate.type.CollectionType;
 import org.hibernate.type.Type;
 import org.hibernate.type.TypeHelper;
@@ -82,49 +87,94 @@ public class DefaultDeleteEventListener implements DeleteEventListener,	Callback
 	 *
 	 */
 	public void onDelete(DeleteEvent event, DeleteContext transientEntities) throws HibernateException {
+		if ( !optimizeUnloadedDelete( event ) ) {
+			delete( event, transientEntities );
+		}
+	}
+
+	private boolean optimizeUnloadedDelete(DeleteEvent event) {
+		final Object object = event.getObject();
+		if ( object instanceof HibernateProxy ) {
+			HibernateProxy proxy = (HibernateProxy) object;
+			LazyInitializer initializer = proxy.getHibernateLazyInitializer();
+			if ( initializer.isUninitialized() ) {
+				final EventSource source = event.getSession();
+				final EntityPersister persister = source.getFactory().getMappingMetamodel()
+						.findEntityDescriptor( initializer.getEntityName() );
+				final Object id = initializer.getIdentifier();
+				final EntityKey key = source.generateEntityKey( id, persister );
+				final PersistenceContext persistenceContext = source.getPersistenceContextInternal();
+				if ( !persistenceContext.containsEntity( key )
+						&& canBeDeletedWithoutLoading( source, persister ) ) {
+					// optimization for deleting certain entities without loading them
+					persistenceContext.reassociateProxy( object, id );
+					if ( !persistenceContext.containsDeletedUnloadedEntityKey( key ) ) {
+						persistenceContext.registerDeletedUnloadedEntityKey( key );
+						source.getActionQueue().addAction(
+								new EntityDeleteAction(
+										id,
+										null,
+										null,
+										null,
+										persister,
+										false,
+										source
+								)
+						);
+					}
+					return true;
+				}
+			}
+		}
+		return false;
+	}
+
+	private void delete(DeleteEvent event, DeleteContext transientEntities) {
+		final PersistenceContext persistenceContext = event.getSession().getPersistenceContextInternal();
+		final Object entity = persistenceContext.unproxyAndReassociate( event.getObject() );
+		EntityEntry entityEntry = persistenceContext.getEntry( entity );
+		if ( entityEntry == null ) {
+			deleteTransientInstance( event, transientEntities, entity );
+		}
+		else {
+			deletePersistentInstance( event, transientEntities, entity, entityEntry );
+		}
+	}
+
+	private void deleteTransientInstance(
+			DeleteEvent event,
+			DeleteContext transientEntities,
+			Object entity) {
+
+		LOG.trace( "Entity was not persistent in delete processing" );
 
 		final EventSource source = event.getSession();
 
-		final PersistenceContext persistenceContext = source.getPersistenceContextInternal();
+		EntityPersister persister = source.getEntityPersister(event.getEntityName(), entity);
-		Object entity = persistenceContext.unproxyAndReassociate( event.getObject() );
+		if ( ForeignKeys.isTransient( persister.getEntityName(), entity, null, source ) ) {
-
+			deleteTransientEntity( source, entity, event.isCascadeDeleteEnabled(), persister, transientEntities );
-		EntityEntry entityEntry = persistenceContext.getEntry( entity );
+		}
-		final EntityPersister persister;
+		else {
-		final Object id;
-		final Object version;
-
-		if ( entityEntry == null ) {
-			LOG.trace( "Entity was not persistent in delete processing" );
-
-			persister = source.getEntityPersister( event.getEntityName(), entity );
-
-			if ( ForeignKeys.isTransient( persister.getEntityName(), entity, null, source ) ) {
-				deleteTransientEntity( source, entity, event.isCascadeDeleteEnabled(), persister, transientEntities );
-				// EARLY EXIT!!!
-				return;
-			}
 			performDetachedEntityDeletionCheck( event );
 
-			id = persister.getIdentifier( entity, source );
+			final Object id = persister.getIdentifier( entity, source );
-
 			if ( id == null ) {
-				throw new TransientObjectException(
+				throw new TransientObjectException("the detached instance passed to delete() had a null identifier");
-						"the detached instance passed to delete() had a null identifier"
-				);
 			}
 
+			final PersistenceContext persistenceContext = source.getPersistenceContextInternal();
 			final EntityKey key = source.generateEntityKey( id, persister );
 
-			persistenceContext.checkUniqueness( key, entity );
+			persistenceContext.checkUniqueness( key, entity);
 
 			new OnUpdateVisitor( source, id, entity ).process( entity, persister );
 
-			version = persister.getVersion( entity );
+			final Object version = persister.getVersion( entity );
 
-			entityEntry = persistenceContext.addEntity(
+			EntityEntry entityEntry = persistenceContext.addEntity(
 					entity,
 					persister.isMutable() ? Status.MANAGED : Status.READ_ONLY,
-					persister.getValues( entity ),
+					persister.getValues(entity),
 					key,
 					version,
 					LockMode.NONE,
@@ -133,21 +183,51 @@ public class DefaultDeleteEventListener implements DeleteEventListener,	Callback
 					false
 			);
 			persister.afterReassociate( entity, source );
-		}
-		else {
-			LOG.trace( "Deleting a persistent instance" );
 
-			if ( entityEntry.getStatus() == Status.DELETED || entityEntry.getStatus() == Status.GONE ) {
+			delete( event, transientEntities, source, entity, persister, id, version, entityEntry );
-				LOG.trace( "Object was already deleted" );
-				return;
-			}
-			persister = entityEntry.getPersister();
-			id = entityEntry.getId();
-			version = entityEntry.getVersion();
 		}
+	}
 
-		callbackRegistry.preRemove( entity );
+	private void deletePersistentInstance(
-		if ( invokeDeleteLifecycle( source, entity, persister ) ) {
+			DeleteEvent event,
+			DeleteContext transientEntities,
+			Object entity,
+			EntityEntry entityEntry) {
+
+		LOG.trace( "Deleting a persistent instance" );
+
+		final EventSource source = event.getSession();
+
+		if ( entityEntry.getStatus() == Status.DELETED || entityEntry.getStatus() == Status.GONE
+				|| source.getPersistenceContextInternal()
+						.containsDeletedUnloadedEntityKey( entityEntry.getEntityKey() ) ) {
+			LOG.trace( "Object was already deleted" );
+			return;
+		}
+		delete(
+				event,
+				transientEntities,
+				source,
+				entity,
+				entityEntry.getPersister(),
+				entityEntry.getId(),
+				entityEntry.getVersion(),
+				entityEntry
+		);
+	}
+
+	private void delete(
+			DeleteEvent event,
+			DeleteContext transientEntities,
+			EventSource source,
+			Object entity,
+			EntityPersister persister,
+			Object id,
+			Object version,
+			EntityEntry entityEntry) {
+
+		callbackRegistry.preRemove(entity);
+		if ( invokeDeleteLifecycle(source, entity, persister) ) {
 			return;
 		}
 
@@ -162,10 +242,41 @@ public class DefaultDeleteEventListener implements DeleteEventListener,	Callback
 		);
 
 		if ( source.getFactory().getSessionFactoryOptions().isIdentifierRollbackEnabled() ) {
-			persister.resetIdentifier( entity, id, version, source );
+			persister.resetIdentifier(entity, id, version, source);
 		}
 	}
 
+	/**
+	 * Can we delete the row represented by the proxy without loading the entity?
+	 */
+	private boolean canBeDeletedWithoutLoading(EventSource source, EntityPersister persister) {
+		return source.getInterceptor() == EmptyInterceptor.INSTANCE
+			&& !persister.implementsLifecycle()
+			&& !persister.hasSubclasses()
+			&& !persister.hasCascadeDelete()
+			&& !persister.hasOwnedCollections()
+			&& !persister.hasNaturalIdentifier()
+			&& !hasRegisteredRemoveCallbacks( persister )
+			&& !hasCustomEventListeners( source );
+	}
+
+	private static boolean hasCustomEventListeners(EventSource source) {
+		FastSessionServices fss = source.getFactory().getFastSessionServices();
+		// Bean Validation adds a PRE_DELETE listener
+		// and Envers adds a POST_DELETE listener
+		return fss.eventListenerGroup_PRE_DELETE.count() > 0
+			|| fss.eventListenerGroup_POST_DELETE.count() > 1
+			|| fss.eventListenerGroup_POST_DELETE.count() == 1
+				&& !(fss.eventListenerGroup_POST_DELETE.listeners().iterator().next()
+						instanceof PostDeleteEventListenerStandardImpl);
+	}
+
+	private boolean hasRegisteredRemoveCallbacks(EntityPersister persister) {
+		Class<?> mappedClass = persister.getMappedClass();
+		return callbackRegistry.hasRegisteredCallbacks( mappedClass, CallbackType.PRE_REMOVE )
+			|| callbackRegistry.hasRegisteredCallbacks( mappedClass, CallbackType.POST_REMOVE );
+	}
+
 	/**
 	 * Called when we have recognized an attempt to delete a detached entity.
 	 * <p/>

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

@@ -53,5 +53,9 @@ public class DefaultFlushEventListener extends AbstractFlushingEventListener imp
 				statistics.flush();
 			}
 		}
+		else if ( source.getActionQueue().hasAnyQueuedActions() ) {
+			// execute any queued unloaded-entity deletions
+			performExecutions( source );
+		}
 	}
 }

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

@@ -286,8 +286,11 @@ public class DefaultLoadEventListener implements LoadEventListener {
 			// if the entity defines a HibernateProxy factory, see if there is an
 			// existing proxy associated with the PC - and if so, use it
 			if ( persister.getRepresentationStrategy().getProxyFactory() != null ) {
-				final Object proxy = persistenceContext.getProxy( keyToLoad );
+				if ( persistenceContext.containsDeletedUnloadedEntityKey( keyToLoad ) ) {
+					return null;
+				}
 
+				final Object proxy = persistenceContext.getProxy( keyToLoad );
 				if ( proxy != null ) {
 					if( traceEnabled ) {
 						LOG.trace( "Entity proxy found in session cache" );

hibernate-core/src/main/java/org/hibernate/persister/collection/BasicCollectionPersister.java

@@ -198,8 +198,7 @@ public class BasicCollectionPersister extends AbstractCollectionPersister {
 		try {
 			final Expectation expectation = Expectations.appropriateExpectation( getUpdateCheckStyle() );
 			final boolean callable = isUpdateCallable();
-			final int jdbcBatchSizeToUse = session.getConfiguredJdbcBatchSize();
+			boolean useBatch = expectation.canBeBatched() && session.getConfiguredJdbcBatchSize() > 1;
-			boolean useBatch = expectation.canBeBatched() && jdbcBatchSizeToUse > 1;
 			final Iterator<?> entries = collection.entries( this );
 
 			final List<Object> elements = new ArrayList<>();

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

@@ -386,6 +386,7 @@ public abstract class AbstractEntityPersister
 	private String sqlLazyUpdateByRowIdString;
 
 	private String[] sqlDeleteStrings;
+	private String[] sqlDeleteNoVersionCheckStrings;
 	private String[] sqlInsertStrings;
 	private String[] sqlUpdateStrings;
 	private String[] sqlLazyUpdateStrings;
@@ -557,6 +558,10 @@ public abstract class AbstractEntityPersister
 		return sqlDeleteStrings;
 	}
 
+	public String[] getSQLDeleteNoVersionCheckStrings() {
+		return sqlDeleteNoVersionCheckStrings;
+	}
+
 	public String[] getSQLInsertStrings() {
 		return sqlInsertStrings;
 	}
@@ -3146,10 +3151,10 @@ public abstract class AbstractEntityPersister
 	/**
 	 * Generate the SQL that deletes a row by id (and version)
 	 */
-	public String generateDeleteString(int j) {
+	public String generateDeleteString(int j, boolean includeVersion) {
 		final Delete delete = createDelete().setTableName( getTableName( j ) )
 				.addPrimaryKeyColumns( getKeyColumns( j ) );
-		if ( j == 0 ) {
+		if ( includeVersion && j == 0 ) {
 			delete.setVersionColumnName( getVersionColumnName() );
 		}
 		if ( getFactory().getSessionFactoryOptions().isCommentsEnabled() ) {
@@ -3341,11 +3346,9 @@ public abstract class AbstractEntityPersister
 
 		// TODO : shouldn't inserts be Expectations.NONE?
 		final Expectation expectation = Expectations.appropriateExpectation( insertResultCheckStyles[j] );
-		final int jdbcBatchSizeToUse = session.getConfiguredJdbcBatchSize();
+		final boolean useBatch = expectation.canBeBatched()
-		final boolean useBatch = expectation.canBeBatched() &&
+				&& session.getConfiguredJdbcBatchSize() > 1
-						jdbcBatchSizeToUse > 1 &&
+				&& getIdentifierGenerator().supportsJdbcBatchInserts();
-						getIdentifierGenerator().supportsJdbcBatchInserts();
-
 		if ( useBatch && insertBatchKey == null ) {
 			insertBatchKey = new BasicBatchKey(
 					getEntityName() + "#INSERT",
@@ -3483,7 +3486,6 @@ public abstract class AbstractEntityPersister
 			final SharedSessionContractImplementor session) throws HibernateException {
 
 		final Expectation expectation = Expectations.appropriateExpectation( updateResultCheckStyles[j] );
-		final int jdbcBatchSizeToUse = session.getConfiguredJdbcBatchSize();
 		// IMPLEMENTATION NOTE: If Session#saveOrUpdate or #update is used to update an entity, then
 		//                      Hibernate does not have a database snapshot of the existing entity.
 		//                      As a result, oldFields will be null.
@@ -3491,11 +3493,10 @@ public abstract class AbstractEntityPersister
 		// because there is no way to know that there is actually a row to update. If the update
 		// was batched in this case, the batch update would fail and there is no way to fallback to
 		// an insert.
-		final boolean useBatch =
+		final boolean useBatch = expectation.canBeBatched()
-				expectation.canBeBatched() &&
+				&& isBatchable()
-						isBatchable() &&
+				&& session.getConfiguredJdbcBatchSize() > 1
-						jdbcBatchSizeToUse > 1 &&
+				&& ( oldFields != null || !isNullableTable( j ) );
-						( oldFields != null || !isNullableTable( j ) );
 		if ( useBatch && updateBatchKey == null ) {
 			updateBatchKey = new BasicBatchKey(
 					getEntityName() + "#UPDATE",
@@ -3632,10 +3633,14 @@ public abstract class AbstractEntityPersister
 			return;
 		}
 
-		final boolean useVersion = j == 0 && isVersioned();
+		final boolean useVersion = j == 0 && isVersioned()
+				&& object != null; // null object signals that we're deleting an unloaded proxy
 		final boolean callable = isDeleteCallable( j );
 		final Expectation expectation = Expectations.appropriateExpectation( deleteResultCheckStyles[j] );
-		final boolean useBatch = j == 0 && isBatchable() && expectation.canBeBatched();
+		final boolean useBatch = j == 0
+				&& isBatchable()
+				&& expectation.canBeBatched()
+				&& session.getConfiguredJdbcBatchSize() > 1;
 		if ( useBatch && deleteBatchKey == null ) {
 			deleteBatchKey = new BasicBatchKey(
 					getEntityName() + "#DELETE",
@@ -3678,12 +3683,12 @@ public abstract class AbstractEntityPersister
 
 				index += expectation.prepare( delete );
 
-				// Do the key. The key is immutable so we can use the _current_ object state - not necessarily
+				// Do the key. The key is immutable, so we can use the _current_ object state,
-				// the state at the time the delete was issued
+				// not necessarily the state at the time the delete operation was issued
 				getIdentifierType().nullSafeSet( delete, id, index, session );
 				index += getIdentifierColumnSpan();
 
-				// We should use the _current_ object state (ie. after any updates that occurred during flush)
+				// We should use the _current_ object state (after any updates that occurred during flush)
 
 				if ( useVersion ) {
 					getVersionType().nullSafeSet( delete, version, index, session );
@@ -4010,7 +4015,8 @@ public abstract class AbstractEntityPersister
 	public void delete(Object id, Object version, Object object, SharedSessionContractImplementor session)
 			throws HibernateException {
 		final int span = getTableSpan();
-		boolean isImpliedOptimisticLocking = !entityMetamodel.isVersioned() && isAllOrDirtyOptLocking();
+		boolean isImpliedOptimisticLocking = !entityMetamodel.isVersioned() && isAllOrDirtyOptLocking()
+				&& object != null; // null object signals that we're deleting an unloaded proxy
 		Object[] loadedState = null;
 		if ( isImpliedOptimisticLocking ) {
 			// need to treat this as if it where optimistic-lock="all" (dirty does *not* make sense);
@@ -4031,10 +4037,13 @@ public abstract class AbstractEntityPersister
 			// we need to utilize dynamic delete statements
 			deleteStrings = generateSQLDeleteStrings( loadedState );
 		}
-		else {
+		else if (object!=null) {
 			// otherwise, utilize the static delete statements
 			deleteStrings = getSQLDeleteStrings();
 		}
+		else {
+			deleteStrings = getSQLDeleteNoVersionCheckStrings();
+		}
 
 		for ( int j = span - 1; j >= 0; j-- ) {
 			delete( id, version, j, object, deleteStrings[j], session, loadedState );
@@ -4212,6 +4221,7 @@ public abstract class AbstractEntityPersister
 		//insert/update/delete SQL
 		final int joinSpan = getTableSpan();
 		sqlDeleteStrings = new String[joinSpan];
+		sqlDeleteNoVersionCheckStrings = new String[joinSpan];
 		sqlInsertStrings = new String[joinSpan];
 		sqlUpdateStrings = new String[joinSpan];
 		sqlLazyUpdateStrings = new String[joinSpan];
@@ -4226,16 +4236,19 @@ public abstract class AbstractEntityPersister
 		for ( int j = 0; j < joinSpan; j++ ) {
 			sqlInsertStrings[j] = customSQLInsert[j] == null
 					? generateInsertString( getPropertyInsertability(), j )
-					: substituteBrackets( customSQLInsert[j]);
+					: substituteBrackets( customSQLInsert[j] );
 			sqlUpdateStrings[j] = customSQLUpdate[j] == null
 					? generateUpdateString( getPropertyUpdateability(), j, false )
-					: substituteBrackets( customSQLUpdate[j]);
+					: substituteBrackets( customSQLUpdate[j] );
 			sqlLazyUpdateStrings[j] = customSQLUpdate[j] == null
 					? generateUpdateString( getNonLazyPropertyUpdateability(), j, false )
-					: substituteBrackets( customSQLUpdate[j]);
+					: substituteBrackets( customSQLUpdate[j] );
 			sqlDeleteStrings[j] = customSQLDelete[j] == null
-					? generateDeleteString( j )
+					? generateDeleteString( j, true )
-					: substituteBrackets( customSQLDelete[j]);
+					: substituteBrackets( customSQLDelete[j] );
+			sqlDeleteNoVersionCheckStrings[j] = customSQLDelete[j] == null
+					? generateDeleteString( j, false )
+					: substituteBrackets( customSQLDelete[j] ); //TODO: oops, fix!
 		}
 
 		tableHasColumns = new boolean[joinSpan];
@@ -4664,6 +4677,16 @@ public abstract class AbstractEntityPersister
 		return entityMetamodel.hasCascades();
 	}
 
+	@Override
+	public boolean hasCascadeDelete() {
+		return entityMetamodel.hasCascadeDelete();
+	}
+
+	@Override
+	public boolean hasOwnedCollections() {
+		return entityMetamodel.hasOwnedCollections();
+	}
+
 	@Override
 	public boolean hasIdentifierProperty() {
 		return !entityMetamodel.getIdentifierProperty().isVirtual();

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

@@ -280,13 +280,37 @@ public interface EntityPersister
 	boolean hasSubselectLoadableCollections();
 
 	/**
-	 * Determine whether this entity has any non-none cascading.
+	 * Determine whether this entity has any non-{@linkplain org.hibernate.engine.spi.CascadeStyles#NONE none}
+	 * cascading.
 	 *
 	 * @return True if the entity has any properties with a cascade other than NONE;
 	 * false otherwise (aka, no cascading).
 	 */
 	boolean hasCascades();
 
+	/**
+	 * Determine whether this entity has any {@linkplain org.hibernate.engine.spi.CascadeStyles#DELETE delete}
+	 * cascading.
+	 *
+	 * @return True if the entity has any properties with a cascade other than NONE;
+	 * false otherwise.
+	 */
+	default boolean hasCascadeDelete() {
+		//bad default implementation for compatibility
+		return hasCascades();
+	}
+
+	/**
+	 * Determine whether this entity has any owned collections.
+	 *
+	 * @return True if the entity has an owned collection;
+	 * false otherwise.
+	 */
+	default boolean hasOwnedCollections() {
+		//bad default implementation for compatibility
+		return hasCollections();
+	}
+
 	/**
 	 * Determine whether instances of this entity are considered mutable.
 	 *

hibernate-core/src/main/java/org/hibernate/tuple/entity/EntityMetamodel.java

@@ -26,6 +26,7 @@ import org.hibernate.cfg.NotYetImplementedException;
 import org.hibernate.engine.OptimisticLockStyle;
 import org.hibernate.engine.spi.CascadeStyle;
 import org.hibernate.engine.spi.CascadeStyles;
+import org.hibernate.engine.spi.CascadingActions;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.internal.CoreMessageLogger;
 import org.hibernate.internal.util.ReflectHelper;
@@ -47,6 +48,7 @@ import org.hibernate.tuple.PropertyFactory;
 import org.hibernate.tuple.ValueGeneration;
 import org.hibernate.tuple.ValueGenerator;
 import org.hibernate.type.AssociationType;
+import org.hibernate.type.CollectionType;
 import org.hibernate.type.ComponentType;
 import org.hibernate.type.CompositeType;
 import org.hibernate.type.EntityType;
@@ -102,6 +104,7 @@ public class EntityMetamodel implements Serializable {
 
 	private final Map<String, Integer> propertyIndexes = new HashMap<>();
 	private final boolean hasCollections;
+	private final boolean hasOwnedCollections;
 	private final BitSet mutablePropertiesIndexes;
 	private final boolean hasLazyProperties;
 	private final boolean hasNonIdentifierPropertyNamedId;
@@ -112,6 +115,7 @@ public class EntityMetamodel implements Serializable {
 
 	private boolean lazy; //not final because proxy factory creation can fail
 	private final boolean hasCascades;
+	private final boolean hasCascadeDelete;
 	private final boolean mutable;
 	private final boolean isAbstract;
 	private final boolean selectBeforeUpdate;
@@ -214,7 +218,9 @@ public class EntityMetamodel implements Serializable {
 
 		int tempVersionProperty = NO_VERSION_INDX;
 		boolean foundCascade = false;
+		boolean foundCascadeDelete = false;
 		boolean foundCollection = false;
+		boolean foundOwnedCollection = false;
 		BitSet mutableIndexes = new BitSet();
 		boolean foundNonIdentifierPropertyNamedId = false;
 		boolean foundUpdateableNaturalIdProperty = false;
@@ -326,13 +332,19 @@ public class EntityMetamodel implements Serializable {
 				hasLazy = true;
 			}
 
-			if ( attribute.getCascadeStyle() != CascadeStyles.NONE ) {
+			if ( cascadeStyles[i] != CascadeStyles.NONE ) {
 				foundCascade = true;
 			}
+			if ( cascadeStyles[i].doCascade(CascadingActions.DELETE) ) {
+				foundCascadeDelete = true;
+			}
 
 			if ( indicatesCollection( attribute.getType() ) ) {
 				foundCollection = true;
 			}
+			if ( indicatesOwnedCollection( attribute.getType(), creationContext.getMetadata() ) ) {
+				foundOwnedCollection = true;
+			}
 
 			// Component types are dirty tracked as well so they are not exactly mutable for the "maybeDirty" check
 			if ( propertyType.isMutable() && propertyCheckability[i] && !( propertyType instanceof ComponentType ) ) {
@@ -359,6 +371,7 @@ public class EntityMetamodel implements Serializable {
 		this.hasUpdateGeneratedValues = foundPostUpdateGeneratedValues;
 
 		hasCascades = foundCascade;
+		hasCascadeDelete = foundCascadeDelete;
 		hasNonIdentifierPropertyNamedId = foundNonIdentifierPropertyNamedId;
 		versionPropertyIndex = tempVersionProperty;
 		hasLazyProperties = hasLazy;
@@ -409,6 +422,7 @@ public class EntityMetamodel implements Serializable {
 		}
 
 		hasCollections = foundCollection;
+		hasOwnedCollections = foundOwnedCollection;
 		mutablePropertiesIndexes = mutableIndexes;
 
 		final Set<String> subclassEntityNamesLocal = new HashSet<>();
@@ -806,14 +820,30 @@ public class EntityMetamodel implements Serializable {
 		return subclassEntityNames;
 	}
 
-	private boolean indicatesCollection(Type type) {
+	private static boolean indicatesCollection(Type type) {
 		if ( type.isCollectionType() ) {
 			return true;
 		}
 		else if ( type.isComponentType() ) {
 			Type[] subtypes = ( (CompositeType) type ).getSubtypes();
 			for ( Type subtype : subtypes ) {
-					if ( indicatesCollection( subtype ) ) {
+				if ( indicatesCollection( subtype ) ) {
+					return true;
+				}
+			}
+		}
+		return false;
+	}
+
+	private static boolean indicatesOwnedCollection(Type type, MetadataImplementor metadata) {
+		if ( type.isCollectionType() ) {
+			String role = ( (CollectionType) type ).getRole();
+			return !metadata.getCollectionBinding( role ).isInverse();
+		}
+		else if ( type.isComponentType() ) {
+			Type[] subtypes = ( (CompositeType) type ).getSubtypes();
+			for ( Type subtype : subtypes ) {
+				if ( indicatesOwnedCollection( subtype, metadata ) ) {
 					return true;
 				}
 			}
@@ -881,6 +911,10 @@ public class EntityMetamodel implements Serializable {
 		return hasCollections;
 	}
 
+	public boolean hasOwnedCollections() {
+		return hasOwnedCollections;
+	}
+
 	public boolean hasMutableProperties() {
 		return !mutablePropertiesIndexes.isEmpty();
 	}
@@ -901,6 +935,10 @@ public class EntityMetamodel implements Serializable {
 		return hasCascades;
 	}
 
+	public boolean hasCascadeDelete() {
+		return hasCascadeDelete;
+	}
+
 	public boolean isMutable() {
 		return mutable;
 	}

hibernate-core/src/test/java/org/hibernate/orm/test/deleteunloaded/Child.java

@@ -0,0 +1,27 @@
+package org.hibernate.orm.test.deleteunloaded;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import jakarta.persistence.ManyToOne;
+
+@Entity
+public class Child {
+    @GeneratedValue
+    @Id
+    private long id;
+    @ManyToOne
+    private Parent parent;
+
+    public long getId() {
+        return id;
+    }
+
+    public Parent getParent() {
+        return parent;
+    }
+
+    public void setParent(Parent parent) {
+        this.parent = parent;
+    }
+}

hibernate-core/src/test/java/org/hibernate/orm/test/deleteunloaded/DeleteUnloadedProxyTest.java

@@ -0,0 +1,85 @@
+package org.hibernate.orm.test.deleteunloaded;
+
+import org.hibernate.Transaction;
+import org.hibernate.testing.orm.junit.DomainModel;
+import org.hibernate.testing.orm.junit.SessionFactory;
+import org.hibernate.testing.orm.junit.SessionFactoryScope;
+import org.junit.jupiter.api.Test;
+
+import static org.hibernate.Hibernate.isInitialized;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+@DomainModel( annotatedClasses = { Parent.class, Child.class } )
+@SessionFactory
+//@ServiceRegistry(
+//        settings = {
+//                @Setting(name = Environment.STATEMENT_BATCH_SIZE, value = "0")
+//        }
+//)
+public class DeleteUnloadedProxyTest {
+    @Test
+    public void testAttached(SessionFactoryScope scope) {
+        Parent p = new Parent();
+        Child c = new Child();
+        scope.inSession( em -> {
+            Transaction tx = em.beginTransaction();
+            c.setParent(p);
+            p.getChildren().add(c);
+            em.persist(p);
+            tx.commit();
+        } );
+        scope.inSession( em -> {
+            Transaction tx = em.beginTransaction();
+            Child child = em.getReference( Child.class, c.getId() );
+            assertFalse( isInitialized(child) );
+            em.remove(child);
+            Parent parent = em.getReference( Parent.class, p.getId() );
+            assertFalse( isInitialized(parent) );
+            em.remove(parent);
+            tx.commit();
+            assertFalse( isInitialized(child) );
+            assertFalse( isInitialized(parent) );
+        } );
+        scope.inSession( em -> {
+            assertNull( em.find( Parent.class, p.getId() ) );
+            assertNull( em.find( Child.class, c.getId() ) );
+        } );
+    }
+    @Test
+    public void testDetached(SessionFactoryScope scope) {
+        Parent p = new Parent();
+        Child c = new Child();
+        scope.inSession( em -> {
+            Transaction tx = em.beginTransaction();
+            c.setParent(p);
+            p.getChildren().add(c);
+            em.persist(p);
+            tx.commit();
+        } );
+        Child cc = scope.fromSession( em -> {
+            Transaction tx = em.beginTransaction();
+            Child child = em.getReference( Child.class, c.getId() );
+            assertFalse( isInitialized(child) );
+            return child;
+        } );
+        Parent pp = scope.fromSession( em -> {
+            Transaction tx = em.beginTransaction();
+            Parent parent = em.getReference( Parent.class, p.getId() );
+            assertFalse( isInitialized(parent) );
+            return parent;
+        } );
+        scope.inSession( em -> {
+            Transaction tx = em.beginTransaction();
+            em.remove(cc);
+            em.remove(pp);
+            tx.commit();
+            assertFalse( isInitialized(cc) );
+            assertFalse( isInitialized(pp) );
+        } );
+        scope.inSession( em -> {
+            assertNull( em.find( Parent.class, p.getId() ) );
+            assertNull( em.find( Child.class, c.getId() ) );
+        } );
+    }
+}

hibernate-core/src/test/java/org/hibernate/orm/test/deleteunloaded/Parent.java

@@ -0,0 +1,31 @@
+package org.hibernate.orm.test.deleteunloaded;
+
+import jakarta.persistence.CascadeType;
+import jakarta.persistence.Entity;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.Id;
+import jakarta.persistence.OneToMany;
+import jakarta.persistence.Version;
+
+import java.util.HashSet;
+import java.util.Set;
+
+@Entity
+public class Parent {
+    @GeneratedValue
+    @Id
+    private long id;
+    @Version
+    private int version;
+
+    @OneToMany(mappedBy = "parent", cascade = CascadeType.PERSIST)
+    private Set<Child> children = new HashSet<>();
+
+    public Set<Child> getChildren() {
+        return children;
+    }
+
+    public long getId() {
+        return id;
+    }
+}