HHH-19774 - Automatic flushing for child session with shared connection/tx

hibernate-core/src/main/java/org/hibernate/engine/creation/CommonSharedBuilder.java

@@ -30,6 +30,7 @@ public interface CommonSharedBuilder extends CommonBuilder {
 
 	/**
 	 * Signifies that the connection from the original session should be used to create the new session.
+	 * Implies that the overall "transaction context" should be shared as well.
 	 *
 	 * @return {@code this}, for method chaining
 	 */

hibernate-core/src/main/java/org/hibernate/engine/creation/internal/ParentSessionObserver.java

@@ -0,0 +1,27 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.engine.creation.internal;
+
+import org.hibernate.engine.creation.CommonSharedBuilder;
+
+/**
+ * Allows observation of flush and closure events of a parent session from a
+ * child session which shares connection/transaction with the parent.
+ *
+ * @see CommonSharedBuilder#connection()
+ *
+ * @author Steve Ebersole
+ */
+public interface ParentSessionObserver {
+	/**
+	 * Callback when the parent is flushed.  Used to flush the child session.
+	 */
+	void onParentFlush();
+
+	/**
+	 * Callback when the parent is closed.  Used to close the child session.
+	 */
+	void onParentClose();
+}

hibernate-core/src/main/java/org/hibernate/engine/creation/internal/SessionCreationOptionsAdaptor.java

@@ -12,6 +12,7 @@
 import org.hibernate.engine.jdbc.spi.JdbcCoordinator;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.TransactionCompletionCallbacksImplementor;
+import org.hibernate.internal.AbstractSharedSessionContract;
 import org.hibernate.resource.jdbc.spi.PhysicalConnectionHandlingMode;
 import org.hibernate.resource.jdbc.spi.StatementInspector;
 import org.hibernate.resource.transaction.spi.TransactionCoordinator;
@@ -31,7 +32,8 @@
  */
 public record SessionCreationOptionsAdaptor(
 		SessionFactoryImplementor factory,
-		CommonSharedSessionCreationOptions options)
+		CommonSharedSessionCreationOptions options,
+		AbstractSharedSessionContract originalSession)
 			implements SharedSessionCreationOptions {
 
 	@Override
@@ -143,4 +145,9 @@ public Transaction getTransaction() {
 	public TransactionCompletionCallbacksImplementor getTransactionCompletionCallbacks() {
 		return options.getTransactionCompletionCallbacksImplementor();
 	}
+
+	@Override
+	public void registerParentSessionObserver(ParentSessionObserver observer) {
+		registerParentSessionObserver( observer, originalSession );
+	}
 }

hibernate-core/src/main/java/org/hibernate/engine/creation/internal/SharedSessionBuilderImpl.java

@@ -100,6 +100,7 @@ public SessionImplementor openSession() {
 
 	@Override
 	@Deprecated(forRemoval = true)
+	@SuppressWarnings("removal")
 	public SharedSessionBuilderImplementor tenantIdentifier(String tenantIdentifier) {
 		tenantIdentifier( (Object) tenantIdentifier );
 		return this;
@@ -259,6 +260,11 @@ public SharedSessionBuilderImplementor subselectFetchEnabled(boolean subselectFe
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// SharedSessionCreationOptions
 
+	@Override
+	public void registerParentSessionObserver(ParentSessionObserver observer) {
+		registerParentSessionObserver( observer, original );
+	}
+
 	@Override
 	public boolean isTransactionCoordinatorShared() {
 		return shareTransactionContext;

hibernate-core/src/main/java/org/hibernate/engine/creation/internal/SharedSessionCreationOptions.java

@@ -4,8 +4,10 @@
  */
 package org.hibernate.engine.creation.internal;
 
+import org.hibernate.SessionEventListener;
 import org.hibernate.Transaction;
 import org.hibernate.engine.jdbc.spi.JdbcCoordinator;
+import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.engine.spi.TransactionCompletionCallbacksImplementor;
 import org.hibernate.resource.transaction.spi.TransactionCoordinator;
 
@@ -26,4 +28,32 @@ public interface SharedSessionCreationOptions extends SessionCreationOptions {
 	JdbcCoordinator getJdbcCoordinator();
 	Transaction getTransaction();
 	TransactionCompletionCallbacksImplementor getTransactionCompletionCallbacks();
+
+	/**
+	 * Registers callbacks for the child session to integrate with events of the parent session.
+	 */
+	void registerParentSessionObserver(ParentSessionObserver observer);
+
+	/**
+	 * Consolidated implementation of adding the parent session observer.
+	 */
+	default void registerParentSessionObserver(ParentSessionObserver observer, SharedSessionContractImplementor original) {
+		original.getEventListenerManager().addListener( new SessionEventListener() {
+			@Override
+			public void flushEnd(int numberOfEntities, int numberOfCollections) {
+				observer.onParentFlush();
+			}
+
+			@Override
+			public void partialFlushEnd(int numberOfEntities, int numberOfCollections) {
+				observer.onParentFlush();
+			}
+
+			@Override
+			public void end() {
+				observer.onParentClose();
+			}
+		} );
+	}
+
 }

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

@@ -24,38 +24,40 @@ public class DefaultFlushEventListener extends AbstractFlushingEventListener imp
 	 */
 	public void onFlush(FlushEvent event) throws HibernateException {
 		final var source = event.getSession();
-		final var persistenceContext = source.getPersistenceContextInternal();
+
 		final var eventMonitor = source.getEventMonitor();
-		if ( persistenceContext.getNumberOfManagedEntities() > 0
-				|| persistenceContext.getCollectionEntriesSize() > 0 ) {
-			EVENT_LISTENER_LOGGER.executingFlush();
-			final var flushEvent = eventMonitor.beginFlushEvent();
-			final var eventListenerManager = source.getEventListenerManager();
-			try {
-				eventListenerManager.flushStart();
+		final var flushEvent = eventMonitor.beginFlushEvent();
+
+		final var eventListenerManager = source.getEventListenerManager();
+		eventListenerManager.flushStart();
+
+		try {
+			final var persistenceContext = source.getPersistenceContextInternal();
+			if ( persistenceContext.getNumberOfManagedEntities() > 0
+					|| persistenceContext.getCollectionEntriesSize() > 0 ) {
+				EVENT_LISTENER_LOGGER.executingFlush();
 				flushEverythingToExecutions( event );
 				performExecutions( source );
 				postFlush( source );
-			}
-			finally {
-				eventMonitor.completeFlushEvent( flushEvent, event );
-				eventListenerManager.flushEnd(
-						event.getNumberOfEntitiesProcessed(),
-						event.getNumberOfCollectionsProcessed()
-				);
-			}
-
-			postPostFlush( source );
+				postPostFlush( source );
 
-			final var statistics = source.getFactory().getStatistics();
-			if ( statistics.isStatisticsEnabled() ) {
-				statistics.flush();
+				final var statistics = source.getFactory().getStatistics();
+				if ( statistics.isStatisticsEnabled() ) {
+					statistics.flush();
+				}
+			}
+			else if ( source.getActionQueue().hasAnyQueuedActions() ) {
+				EVENT_LISTENER_LOGGER.executingFlush();
+				// execute any queued unloaded-entity deletions
+				performExecutions( source );
 			}
 		}
-		else if ( source.getActionQueue().hasAnyQueuedActions() ) {
-			EVENT_LISTENER_LOGGER.executingFlush();
-			// execute any queued unloaded-entity deletions
-			performExecutions( source );
+		finally {
+			eventMonitor.completeFlushEvent( flushEvent, event );
+			eventListenerManager.flushEnd(
+					event.getNumberOfEntitiesProcessed(),
+					event.getNumberOfCollectionsProcessed()
+			);
 		}
 	}
 }

hibernate-core/src/main/java/org/hibernate/internal/AbstractSharedSessionContract.java

@@ -10,7 +10,6 @@
 import jakarta.persistence.criteria.CriteriaDelete;
 import jakarta.persistence.criteria.CriteriaQuery;
 import jakarta.persistence.criteria.CriteriaUpdate;
-
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.hibernate.CacheMode;
 import org.hibernate.EntityNameResolver;
@@ -28,6 +27,7 @@
 import org.hibernate.bytecode.enhance.spi.interceptor.SessionAssociationMarkers;
 import org.hibernate.cache.spi.CacheTransactionSynchronization;
 import org.hibernate.dialect.Dialect;
+import org.hibernate.engine.creation.internal.ParentSessionObserver;
 import org.hibernate.engine.creation.internal.SessionCreationOptions;
 import org.hibernate.engine.creation.internal.SessionCreationOptionsAdaptor;
 import org.hibernate.engine.creation.internal.SharedSessionBuilderImpl;
@@ -225,6 +225,17 @@ public AbstractSharedSessionContract(SessionFactoryImpl factory, SessionCreation
 			jdbcSessionContext = createJdbcSessionContext( statementInspector );
 			logInconsistentOptions( sharedOptions );
 			addSharedSessionTransactionObserver( transactionCoordinator );
+			sharedOptions.registerParentSessionObserver( new ParentSessionObserver() {
+				@Override
+				public void onParentFlush() {
+					propagateFlush();
+				}
+
+				@Override
+				public void onParentClose() {
+					propagateClose();
+				}
+			} );
 		}
 		else {
 			isTransactionCoordinatorShared = false;
@@ -249,7 +260,7 @@ public SharedStatelessSessionBuilder statelessWithOptions() {
 			@Override
 			protected StatelessSessionImplementor createStatelessSession() {
 				return new StatelessSessionImpl( factory,
-						new SessionCreationOptionsAdaptor( factory, this ) );
+						new SessionCreationOptionsAdaptor( factory, this, AbstractSharedSessionContract.this ) );
 			}
 		};
 	}
@@ -538,6 +549,10 @@ protected void setClosed() {
 		cleanupOnClose();
 	}
 
+	protected abstract void propagateFlush();
+
+	protected abstract void propagateClose();
+
 	protected void checkBeforeClosingJdbcCoordinator() {
 	}
 
@@ -893,7 +908,7 @@ public void setNativeJdbcParametersIgnored(boolean nativeJdbcParametersIgnored)
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// dynamic HQL handling
 
-	@Override @SuppressWarnings("rawtypes")
+	@Override @Deprecated @SuppressWarnings({"rawtypes", "deprecation"})
 	public QueryImplementor createQuery(String queryString) {
 		return createQuery( queryString, null );
 	}
@@ -995,10 +1010,12 @@ public <R> QueryImplementor<R> createQuery(TypedQueryReference<R> typedQueryRefe
 		checksBeforeQueryCreation();
 		if ( typedQueryReference instanceof SelectionSpecificationImpl<R> specification ) {
 			final var query = specification.buildCriteria( getCriteriaBuilder() );
+			//noinspection unchecked
 			return new SqmQueryImpl<>( (SqmStatement<R>) query, specification.getResultType(), this );
 		}
 		else if ( typedQueryReference instanceof MutationSpecificationImpl<?> specification ) {
 			final var query = specification.buildCriteria( getCriteriaBuilder() );
+			//noinspection unchecked
 			return new SqmQueryImpl<>( (SqmStatement<R>) query, (Class<R>) specification.getResultType(), this );
 		}
 		else {
@@ -1017,12 +1034,12 @@ else if ( typedQueryReference instanceof MutationSpecificationImpl<?> specificat
 	// ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 	// dynamic native (SQL) query handling
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public NativeQueryImplementor<?> createNativeQuery(String sqlString) {
 		return createNativeQuery( sqlString, (Class<?>) null );
 	}
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public NativeQueryImplementor<?> createNativeQuery(String sqlString, String resultSetMappingName) {
 		checksBeforeQueryCreation();
 		return buildNativeQuery( sqlString, resultSetMappingName, null );
@@ -1124,7 +1141,7 @@ public QueryImplementor<?> getNamedQuery(String queryName) {
 		return createNamedQuery( queryName );
 	}
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public QueryImplementor<?> createNamedQuery(String name) {
 		checksBeforeQueryCreation();
 		try {
@@ -1231,12 +1248,14 @@ protected <T,Q extends CommonQueryContract> Q buildNamedQuery(
 		// first see if it is a named HQL query
 		final var namedSqmQueryMemento = getSqmQueryMemento( queryName );
 		if ( namedSqmQueryMemento != null ) {
+			//noinspection unchecked
 			return sqmCreator.apply( (NamedSqmQueryMemento<T>) namedSqmQueryMemento );
 		}
 
 		// otherwise, see if it is a named native query
 		final var namedNativeDescriptor = getNativeQueryMemento( queryName );
 		if ( namedNativeDescriptor != null ) {
+			//noinspection unchecked
 			return nativeCreator.apply( (NamedNativeQueryMemento<T>) namedNativeDescriptor );
 		}
 
@@ -1300,7 +1319,7 @@ protected <T> SqmQueryImplementor<T> createSqmQueryImplementor(Class<T> resultTy
 		return query;
 	}
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public NativeQueryImplementor<?> getNamedNativeQuery(String queryName) {
 		final var namedNativeDescriptor = getNativeQueryMemento( queryName );
 		if ( namedNativeDescriptor != null ) {
@@ -1311,7 +1330,7 @@ public NativeQueryImplementor<?> getNamedNativeQuery(String queryName) {
 		}
 	}
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public NativeQueryImplementor<?> getNamedNativeQuery(String queryName, String resultSetMapping) {
 		final var namedNativeDescriptor = getNativeQueryMemento( queryName );
 		if ( namedNativeDescriptor != null ) {
@@ -1568,7 +1587,7 @@ public <T> QueryImplementor<T> createQuery(CriteriaQuery<T> criteriaQuery) {
 		}
 	}
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public QueryImplementor<?> createQuery(@SuppressWarnings("rawtypes") CriteriaUpdate criteriaUpdate) {
 		checkOpen();
 		try {
@@ -1580,7 +1599,7 @@ public QueryImplementor<?> createQuery(@SuppressWarnings("rawtypes") CriteriaUpd
 		}
 	}
 
-	@Override
+	@Override @Deprecated @SuppressWarnings("deprecation")
 	public QueryImplementor<?> createQuery(@SuppressWarnings("rawtypes") CriteriaDelete criteriaDelete) {
 		checkOpen();
 		try {