diff --git a/documentation/src/main/asciidoc/introduction/Advanced.adoc b/documentation/src/main/asciidoc/introduction/Advanced.adoc
index 26c74a8e658b..544ea7714cf1 100644
--- a/documentation/src/main/asciidoc/introduction/Advanced.adoc
+++ b/documentation/src/main/asciidoc/introduction/Advanced.adoc
@@ -268,16 +268,34 @@ var entityManager =
 
 However, since we often don't have this level of control over creation of the session, it's more common to supply an implementation of link:{doc-javadoc-url}org/hibernate/context/spi/CurrentTenantIdentifierResolver.html[`CurrentTenantIdentifierResolver`] to Hibernate.
 
-There are three common ways to implement multi-tenancy:
+To make use of multi-tenancy, we'll usually need to set at least one of these configuration properties:
+
+.Multi-tenancy configuration
+[%breakable,cols="36,~"]
+|===
+| Configuration property name           | Purpose
+
+| link:{doc-javadoc-url}org/hibernate/cfg/MultiTenancySettings.html#MULTI_TENANT_IDENTIFIER_RESOLVER[`hibernate.tenant_identifier_resolver`]  | Specifies the `CurrentTenantIdentifierResolver`
+| link:{doc-javadoc-url}org/hibernate/cfg/MultiTenancySettings.html#MULTI_TENANT_SCHEMA_MAPPER[`hibernate.multi_tenant.schema_mapper`]  | Specifies the `TenantSchemaMapper` for schema-based multi-tenancy
+| link:{doc-javadoc-url}org/hibernate/cfg/MultiTenancySettings.html#MULTI_TENANT_CONNECTION_PROVIDER[`hibernate.multi_tenant_connection_provider`]  | Specifies the `MultiTenantConnectionProvider` for database-based multi-tenancy
+|===
+
+// Do not configure those properties if you would like the configured `BeanContainer` provide the implementation.
+
+There are three common approaches to multi-tenancy:
 
 1. each tenant has its own database,
 2. each tenant has its own schema, or
 3. tenants share tables in a single schema, and rows are tagged with the tenant id.
 
-From the point of view of Hibernate, there's little difference between the first two options.
-Hibernate will need to obtain a JDBC connection with permissions on the database and schema owned by the current tenant.
+[discrete]
+==== Database-based multi-tenancy
 
-Therefore, we must implement a link:{doc-javadoc-url}org/hibernate/engine/jdbc/connections/spi/MultiTenantConnectionProvider.html[`MultiTenantConnectionProvider`] which takes on this responsibility:
+The first option is to give each tenant its own database.
+That is, we'll use a separate source of JDBC connections for each tenant.
+
+The interface link:{doc-javadoc-url}org/hibernate/engine/jdbc/connections/spi/MultiTenantConnectionProvider.html[`MultiTenantConnectionProvider`] is responsible for obtaining an appropriate `Connection` for a given tenant.
+Typically, we'll provide a custom implementation of this interface:
 
 - from time to time, Hibernate will ask for a connection, passing the id of the current tenant, and then we must create an appropriate connection or obtain one from a pool, and return it to Hibernate, and
 - later, Hibernate will release the connection and ask us to destroy it or return it to the appropriate pool.
@@ -285,10 +303,21 @@ Therefore, we must implement a link:{doc-javadoc-url}org/hibernate/engine/jdbc/c
 [TIP]
 ====
 Check out link:{doc-javadoc-url}org/hibernate/engine/jdbc/connections/spi/DataSourceBasedMultiTenantConnectionProviderImpl.html[`DataSourceBasedMultiTenantConnectionProviderImpl`] for inspiration.
+If your source of JDBC connections is a set of JNDI-bound ``DataSource``s, you might even be able to use this implementation directly.
 ====
 
+[discrete]
+==== Schema-based multi-tenancy
+
+The second option is to keep all the data for different tenants in the same database, giving each tenant a different named database schema with its own set of tables.
+
+In this case we must supply a link:{doc-javadoc-url}org/hibernate/context/spi/TenantSchemaMapper.html[`TenantSchemaMapper`] which is responsible for mapping from tenant ids to schema names.
+
+[discrete]
+==== Discriminator-based multi-tenancy
+
 The third option is quite different.
-In this case we don't need a `MultiTenantConnectionProvider`, but we will need a dedicated column holding the tenant id mapped by each of our entities.
+In this case we store data from all tenants in the same tables, but each table has a dedicated column holding the tenant id mapped by each of our entities.
 
 [source,java]
 ----
@@ -309,20 +338,6 @@ Within a given session, our data is automatically filtered so that only rows tag
 Native SQL queries are _not_ automatically filtered by tenant id; you'll have to do that part yourself.
 ====
 
-To make use of multi-tenancy, we'll usually need to set at least one of these configuration properties:
-
-.Multi-tenancy configuration
-[%breakable,cols="36,~"]
-|===
-| Configuration property name           | Purpose
-
-| link:{doc-javadoc-url}org/hibernate/cfg/MultiTenancySettings.html#MULTI_TENANT_IDENTIFIER_RESOLVER[`hibernate.tenant_identifier_resolver`]  | Specifies the `CurrentTenantIdentifierResolver`
-| link:{doc-javadoc-url}org/hibernate/cfg/MultiTenancySettings.html#MULTI_TENANT_CONNECTION_PROVIDER[`hibernate.multi_tenant_connection_provider`]  | Specifies the `MultiTenantConnectionProvider`
-|===
-
-Do not configure those properties if you would like the configured `BeanContainer` provide the implementation.
-A longer discussion of multi-tenancy may be found in the {multitenacy-doc}[User Guide].
-
 [[custom-sql]]
 === Using custom-written SQL
 
diff --git a/hibernate-core/src/main/java/org/hibernate/boot/SessionFactoryBuilder.java b/hibernate-core/src/main/java/org/hibernate/boot/SessionFactoryBuilder.java
index 7ca0666576db..0d86037ae7f4 100644
--- a/hibernate-core/src/main/java/org/hibernate/boot/SessionFactoryBuilder.java
+++ b/hibernate-core/src/main/java/org/hibernate/boot/SessionFactoryBuilder.java
@@ -15,6 +15,7 @@
 import org.hibernate.annotations.CacheLayout;
 import org.hibernate.cache.spi.TimestampsCacheFactory;
 import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.jpa.spi.JpaCompliance;
 import org.hibernate.proxy.EntityNotFoundDelegate;
 import org.hibernate.query.sqm.function.SqmFunctionDescriptor;
@@ -384,6 +385,21 @@ public interface SessionFactoryBuilder {
 	 */
 	SessionFactoryBuilder applyCurrentTenantIdentifierResolver(CurrentTenantIdentifierResolver<?> resolver);
 
+	/**
+	 * Specifies a {@link TenantSchemaMapper} that is responsible for
+	 * mapping the current tenant identifier to the name of a database
+	 * schema.
+	 *
+	 * @param mapper The mapping strategy to use.
+	 *
+	 * @return {@code this}, for method chaining
+	 *
+	 * @see org.hibernate.cfg.AvailableSettings#MULTI_TENANT_SCHEMA_MAPPER
+	 *
+	 * @since 7.1
+	 */
+	SessionFactoryBuilder applyTenantSchemaMapper(TenantSchemaMapper<?> mapper);
+
 	/**
 	 * If using the built-in JTA-based
 	 * {@link org.hibernate.resource.transaction.spi.TransactionCoordinator} or
diff --git a/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryBuilderImpl.java b/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryBuilderImpl.java
index 762358c00be2..c73391ebe3a4 100644
--- a/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryBuilderImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryBuilderImpl.java
@@ -23,6 +23,7 @@
 import org.hibernate.bytecode.spi.BytecodeProvider;
 import org.hibernate.cache.spi.TimestampsCacheFactory;
 import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.internal.SessionFactoryImpl;
 import org.hibernate.proxy.EntityNotFoundDelegate;
 import org.hibernate.query.sqm.function.SqmFunctionDescriptor;
@@ -252,6 +253,12 @@ public SessionFactoryBuilder applyCurrentTenantIdentifierResolver(CurrentTenantI
 		return this;
 	}
 
+	@Override
+	public SessionFactoryBuilder applyTenantSchemaMapper(TenantSchemaMapper<?> mapper) {
+		this.optionsBuilder.applyTenantSchemaMapper( mapper );
+		return this;
+	}
+
 	@Override
 	public SessionFactoryBuilder applyNamedQueryCheckingOnStartup(boolean enabled) {
 		this.optionsBuilder.enableNamedQueryCheckingOnStartup( enabled );
diff --git a/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryOptionsBuilder.java b/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryOptionsBuilder.java
index 642449a502fa..5473dc091fc9 100644
--- a/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryOptionsBuilder.java
+++ b/hibernate-core/src/main/java/org/hibernate/boot/internal/SessionFactoryOptionsBuilder.java
@@ -28,6 +28,7 @@
 import org.hibernate.LockOptions;
 import org.hibernate.SessionEventListener;
 import org.hibernate.SessionFactoryObserver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.type.TimeZoneStorageStrategy;
 import org.hibernate.annotations.CacheLayout;
 import org.hibernate.boot.SchemaAutoTooling;
@@ -187,6 +188,7 @@ public class SessionFactoryOptionsBuilder implements SessionFactoryOptions {
 	// multi-tenancy
 	private boolean multiTenancyEnabled;
 	private CurrentTenantIdentifierResolver<Object> currentTenantIdentifierResolver;
+	private TenantSchemaMapper<Object> tenantSchemaMapper;
 
 	// Queries
 	private SqmFunctionRegistry sqmFunctionRegistry;
@@ -371,6 +373,9 @@ public SessionFactoryOptionsBuilder(StandardServiceRegistry serviceRegistry, Boo
 				null
 			);
 		}
+		tenantSchemaMapper =
+				strategySelector.resolveStrategy( TenantSchemaMapper.class,
+						settings.get( MULTI_TENANT_SCHEMA_MAPPER ) );
 
 		delayBatchFetchLoaderCreations =
 				configurationService.getSetting( DELAY_ENTITY_LOADER_CREATIONS, BOOLEAN, true );
@@ -1003,6 +1008,11 @@ public boolean isMultiTenancyEnabled() {
 		return multiTenancyEnabled;
 	}
 
+	@Override
+	public TenantSchemaMapper<Object> getTenantSchemaMapper() {
+		return tenantSchemaMapper;
+	}
+
 	@Override
 	public CurrentTenantIdentifierResolver<Object> getCurrentTenantIdentifierResolver() {
 		return currentTenantIdentifierResolver;
@@ -1450,6 +1460,11 @@ public void applyCurrentTenantIdentifierResolver(CurrentTenantIdentifierResolver
 		this.currentTenantIdentifierResolver = (CurrentTenantIdentifierResolver<Object>) resolver;
 	}
 
+	public void applyTenantSchemaMapper(TenantSchemaMapper<?> mapper) {
+		//noinspection unchecked
+		this.tenantSchemaMapper = (TenantSchemaMapper<Object>) mapper;
+	}
+
 	public void enableNamedQueryCheckingOnStartup(boolean enabled) {
 		this.namedQueryStartupCheckingEnabled = enabled;
 	}
diff --git a/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryBuilder.java b/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryBuilder.java
index 657d62ba8b3c..3f82b67a1177 100644
--- a/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryBuilder.java
+++ b/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryBuilder.java
@@ -16,6 +16,7 @@
 import org.hibernate.boot.TempTableDdlTransactionHandling;
 import org.hibernate.cache.spi.TimestampsCacheFactory;
 import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.proxy.EntityNotFoundDelegate;
 import org.hibernate.query.sqm.function.SqmFunctionDescriptor;
 import org.hibernate.resource.jdbc.spi.PhysicalConnectionHandlingMode;
@@ -208,6 +209,12 @@ public T applyCurrentTenantIdentifierResolver(CurrentTenantIdentifierResolver<?>
 		return getThis();
 	}
 
+	@Override
+	public SessionFactoryBuilder applyTenantSchemaMapper(TenantSchemaMapper<?> mapper) {
+		delegate.applyTenantSchemaMapper( mapper );
+		return getThis();
+	}
+
 	@Override
 	public T applyJtaTrackingByThread(boolean enabled) {
 		delegate.applyJtaTrackingByThread( enabled );
diff --git a/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryOptions.java b/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryOptions.java
index 6b86f67d8387..3ec439e03428 100644
--- a/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryOptions.java
+++ b/hibernate-core/src/main/java/org/hibernate/boot/spi/AbstractDelegatingSessionFactoryOptions.java
@@ -17,6 +17,7 @@
 import org.hibernate.Interceptor;
 import org.hibernate.LockOptions;
 import org.hibernate.SessionFactoryObserver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.type.TimeZoneStorageStrategy;
 import org.hibernate.annotations.CacheLayout;
 import org.hibernate.boot.SchemaAutoTooling;
@@ -224,6 +225,11 @@ public CurrentTenantIdentifierResolver<Object> getCurrentTenantIdentifierResolve
 		return delegate.getCurrentTenantIdentifierResolver();
 	}
 
+	@Override
+	public TenantSchemaMapper<Object> getTenantSchemaMapper() {
+		return delegate.getTenantSchemaMapper();
+	}
+
 	@Override
 	public JavaType<Object> getDefaultTenantIdentifierJavaType() {
 		return delegate.getDefaultTenantIdentifierJavaType();
diff --git a/hibernate-core/src/main/java/org/hibernate/boot/spi/SessionFactoryOptions.java b/hibernate-core/src/main/java/org/hibernate/boot/spi/SessionFactoryOptions.java
index 0ae58214f235..63ae89317a6d 100644
--- a/hibernate-core/src/main/java/org/hibernate/boot/spi/SessionFactoryOptions.java
+++ b/hibernate-core/src/main/java/org/hibernate/boot/spi/SessionFactoryOptions.java
@@ -20,6 +20,7 @@
 import org.hibernate.LockOptions;
 import org.hibernate.SessionEventListener;
 import org.hibernate.SessionFactoryObserver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.type.TimeZoneStorageStrategy;
 import org.hibernate.annotations.CacheLayout;
 import org.hibernate.boot.SchemaAutoTooling;
@@ -313,6 +314,18 @@ default SessionEventListener[] buildSessionEventListeners() {
 	 */
 	CurrentTenantIdentifierResolver<Object> getCurrentTenantIdentifierResolver();
 
+	/**
+	 * Obtain a reference to the current {@linkplain TenantSchemaMapper tenant schema mapper},
+	 * which is used to {@linkplain java.sql.Connection#setSchema set the schema} to the
+	 * {@linkplain TenantSchemaMapper#schemaName schema belonging to the current tenant}
+	 * each time a connection is obtained.
+	 *
+	 * @see org.hibernate.cfg.MultiTenancySettings#MULTI_TENANT_SCHEMA_MAPPER
+	 *
+	 * @since 7.1
+	 */
+	TenantSchemaMapper<Object> getTenantSchemaMapper();
+
 	/**
 	 * @see org.hibernate.cfg.TransactionSettings#JTA_TRACK_BY_THREAD
 	 */
diff --git a/hibernate-core/src/main/java/org/hibernate/cfg/Configuration.java b/hibernate-core/src/main/java/org/hibernate/cfg/Configuration.java
index 30bac56e51ae..9bfc079ae81e 100644
--- a/hibernate-core/src/main/java/org/hibernate/cfg/Configuration.java
+++ b/hibernate-core/src/main/java/org/hibernate/cfg/Configuration.java
@@ -48,6 +48,7 @@
 import org.hibernate.boot.registry.StandardServiceRegistryBuilder;
 import org.hibernate.boot.spi.XmlMappingBinderAccess;
 import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
+import org.hibernate.context.spi.TenantSchemaMapper;
 import org.hibernate.internal.CoreLogging;
 import org.hibernate.internal.CoreMessageLogger;
 import org.hibernate.internal.EmptyInterceptor;
@@ -173,7 +174,8 @@ public class Configuration {
 	private EntityNotFoundDelegate entityNotFoundDelegate;
 	private SessionFactoryObserver sessionFactoryObserver;
 	private StatementInspector statementInspector;
-	private CurrentTenantIdentifierResolver<Object> currentTenantIdentifierResolver;
+	private CurrentTenantIdentifierResolver<?> currentTenantIdentifierResolver;
+	private TenantSchemaMapper<?> tenantSchemaMapper;
 	private CustomEntityDirtinessStrategy customEntityDirtinessStrategy;
 	private ColumnOrderingStrategy columnOrderingStrategy;
 	private SharedCacheMode sharedCacheMode;
@@ -939,7 +941,7 @@ public Configuration setStatementInspector(StatementInspector statementInspector
 	/**
 	 * The {@link CurrentTenantIdentifierResolver}, if any, that was added to this configuration.
 	 */
-	public CurrentTenantIdentifierResolver<Object> getCurrentTenantIdentifierResolver() {
+	public CurrentTenantIdentifierResolver<?> getCurrentTenantIdentifierResolver() {
 		return currentTenantIdentifierResolver;
 	}
 
@@ -948,11 +950,32 @@ public CurrentTenantIdentifierResolver<Object> getCurrentTenantIdentifierResolve
 	 *
 	 * @return {@code this} for method chaining
 	 */
-	public Configuration setCurrentTenantIdentifierResolver(CurrentTenantIdentifierResolver<Object> currentTenantIdentifierResolver) {
+	public Configuration setCurrentTenantIdentifierResolver(CurrentTenantIdentifierResolver<?> currentTenantIdentifierResolver) {
 		this.currentTenantIdentifierResolver = currentTenantIdentifierResolver;
 		return this;
 	}
 
+	/**
+	 * The {@link TenantSchemaMapper}, if any, that was added to this configuration.
+	 *
+	 * @since 7.1
+	 */
+	public TenantSchemaMapper<?> getTenantSchemaMapper() {
+		return tenantSchemaMapper;
+	}
+
+	/**
+	 * Specify a {@link TenantSchemaMapper} to be added to this configuration.
+	 *
+	 * @return {@code this} for method chaining
+	 *
+	 * @since 7.1
+	 */
+	public Configuration setTenantSchemaMapper(TenantSchemaMapper<?> tenantSchemaMapper) {
+		this.tenantSchemaMapper = tenantSchemaMapper;
+		return this;
+	}
+
 	/**
 	 * The {@link CustomEntityDirtinessStrategy}, if any, that was added to this configuration.
 	 */
@@ -1082,6 +1105,10 @@ public SessionFactory buildSessionFactory(ServiceRegistry serviceRegistry) throw
 			sessionFactoryBuilder.applyCurrentTenantIdentifierResolver( currentTenantIdentifierResolver );
 		}
 
+		if ( tenantSchemaMapper != null ) {
+			sessionFactoryBuilder.applyTenantSchemaMapper( tenantSchemaMapper );
+		}
+
 		if ( customEntityDirtinessStrategy != null ) {
 			sessionFactoryBuilder.applyCustomEntityDirtinessStrategy( customEntityDirtinessStrategy );
 		}
diff --git a/hibernate-core/src/main/java/org/hibernate/cfg/MultiTenancySettings.java b/hibernate-core/src/main/java/org/hibernate/cfg/MultiTenancySettings.java
index d8d05486676c..70e4a16d21ab 100644
--- a/hibernate-core/src/main/java/org/hibernate/cfg/MultiTenancySettings.java
+++ b/hibernate-core/src/main/java/org/hibernate/cfg/MultiTenancySettings.java
@@ -22,24 +22,44 @@ public interface MultiTenancySettings {
 	String MULTI_TENANT_CONNECTION_PROVIDER = "hibernate.multi_tenant_connection_provider";
 
 	/**
-	 * Specifies a {@link CurrentTenantIdentifierResolver} to use,
-	 * either:
+	 * Specifies a {@link CurrentTenantIdentifierResolver} to use, either:
 	 * <ul>
 	 *     <li>an instance of {@code CurrentTenantIdentifierResolver},
-	 *     <li>a {@link Class} representing an class that implements {@code CurrentTenantIdentifierResolver}, or
+	 *     <li>a {@link Class} representing a class that implements {@code CurrentTenantIdentifierResolver}, or
 	 *     <li>the name of a class that implements {@code CurrentTenantIdentifierResolver}.
 	 * </ul>
 	 *
-	 * @see org.hibernate.boot.SessionFactoryBuilder#applyCurrentTenantIdentifierResolver(CurrentTenantIdentifierResolver)
+	 * @see CurrentTenantIdentifierResolver
+	 * @see org.hibernate.boot.SessionFactoryBuilder#applyCurrentTenantIdentifierResolver
 	 *
 	 * @since 4.1
 	 */
 	String MULTI_TENANT_IDENTIFIER_RESOLVER = "hibernate.tenant_identifier_resolver";
 
 	/**
-	 * During bootstrap, Hibernate needs access to any Connection for access to {@link java.sql.DatabaseMetaData}.
-	 * <p/>
-	 * This setting configures the name of the DataSource to use for this access
+	 * During bootstrap, Hibernate needs access to a {@code Connection} for access
+	 * to the {@link java.sql.DatabaseMetaData}. This setting configures the tenant id
+	 * to use when obtaining the {@link javax.sql.DataSource} to use for this access.
 	 */
 	String TENANT_IDENTIFIER_TO_USE_FOR_ANY_KEY = "hibernate.multi_tenant.datasource.identifier_for_any";
+
+	/**
+	 * Specifies a {@link org.hibernate.context.spi.TenantSchemaMapper} to use, either:
+	 * <ul>
+	 *     <li>an instance of {@code TenantSchemaMapper},
+	 *     <li>a {@link Class} representing a class that implements {@code TenantSchemaMapper}, or
+	 *     <li>the name of a class that implements {@code TenantSchemaMapper}.
+	 * </ul>
+	 * When a tenant schema mapper is set, {@link java.sql.Connection#setSchema(String)}}
+	 * is called on newly acquired JDBC connections with the schema name returned by
+	 * {@link org.hibernate.context.spi.TenantSchemaMapper#schemaName}.
+	 * <p>
+	 * By default, there is no tenant schema mapper.
+	 *
+	 * @see org.hibernate.context.spi.TenantSchemaMapper
+	 * @see org.hibernate.boot.SessionFactoryBuilder#applyTenantSchemaMapper
+	 *
+	 * @since 7.1
+	 */
+	String MULTI_TENANT_SCHEMA_MAPPER = "hibernate.multi_tenant.schema_mapper";
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/context/spi/AbstractCurrentSessionContext.java b/hibernate-core/src/main/java/org/hibernate/context/spi/AbstractCurrentSessionContext.java
index 434d4ff2e5ed..6ff3527c67c5 100644
--- a/hibernate-core/src/main/java/org/hibernate/context/spi/AbstractCurrentSessionContext.java
+++ b/hibernate-core/src/main/java/org/hibernate/context/spi/AbstractCurrentSessionContext.java
@@ -34,7 +34,7 @@ public SessionFactoryImplementor factory() {
 
 	protected SessionBuilder baseSessionBuilder() {
 		final SessionBuilderImplementor builder = factory.withOptions();
-		final CurrentTenantIdentifierResolver<Object> resolver = factory.getCurrentTenantIdentifierResolver();
+		final var resolver = factory.getCurrentTenantIdentifierResolver();
 		if ( resolver != null ) {
 			builder.tenantIdentifier( resolver.resolveCurrentTenantIdentifier() );
 		}
@@ -42,7 +42,7 @@ protected SessionBuilder baseSessionBuilder() {
 	}
 
 	protected void validateExistingSession(Session existingSession) {
-		final CurrentTenantIdentifierResolver<Object> resolver = factory.getCurrentTenantIdentifierResolver();
+		final var resolver = factory.getCurrentTenantIdentifierResolver();
 		if ( resolver != null && resolver.validateExistingCurrentSessions() ) {
 			final Object currentValue = resolver.resolveCurrentTenantIdentifier();
 			final JavaType<Object> tenantIdentifierJavaType = factory.getTenantIdentifierJavaType();
diff --git a/hibernate-core/src/main/java/org/hibernate/context/spi/CurrentTenantIdentifierResolver.java b/hibernate-core/src/main/java/org/hibernate/context/spi/CurrentTenantIdentifierResolver.java
index e4030f757477..45aa1772e12d 100644
--- a/hibernate-core/src/main/java/org/hibernate/context/spi/CurrentTenantIdentifierResolver.java
+++ b/hibernate-core/src/main/java/org/hibernate/context/spi/CurrentTenantIdentifierResolver.java
@@ -4,6 +4,7 @@
  */
 package org.hibernate.context.spi;
 
+
 /**
  * A callback registered with the {@link org.hibernate.SessionFactory} that is
  * responsible for resolving the current tenant identifier.
diff --git a/hibernate-core/src/main/java/org/hibernate/context/spi/TenantSchemaMapper.java b/hibernate-core/src/main/java/org/hibernate/context/spi/TenantSchemaMapper.java
new file mode 100644
index 000000000000..c14f65d07d91
--- /dev/null
+++ b/hibernate-core/src/main/java/org/hibernate/context/spi/TenantSchemaMapper.java
@@ -0,0 +1,36 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.context.spi;
+
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.hibernate.Incubating;
+
+/**
+ * Obtains the name of a database schema for a given tenant identifier when
+ * {@linkplain org.hibernate.cfg.MultiTenancySettings#MULTI_TENANT_SCHEMA_MAPPER
+ * schema-based multitenancy} is enabled.
+ *
+ * @param <T> The type of the tenant id
+ *
+ * @since 7.1
+ *
+ * @author Gavin King
+ */
+@Incubating
+public interface TenantSchemaMapper<T> {
+	/**
+	 * The name of the database schema for data belonging to the tenant with the
+	 * given identifier.
+	 * <p>
+	 * Called when {@value org.hibernate.cfg.MultiTenancySettings#MULTI_TENANT_SCHEMA_MAPPER}
+	 * is enabled.
+	 *
+	 * @param tenantIdentifier The tenant identifier
+	 * @return The name of the database schema belonging to that tenant
+	 *
+	 * @see org.hibernate.cfg.MultiTenancySettings#MULTI_TENANT_SCHEMA_MAPPER
+	 */
+	@NonNull String schemaName(@NonNull T tenantIdentifier);
+}
diff --git a/hibernate-core/src/main/java/org/hibernate/engine/jdbc/connections/spi/DataSourceBasedMultiTenantConnectionProviderImpl.java b/hibernate-core/src/main/java/org/hibernate/engine/jdbc/connections/spi/DataSourceBasedMultiTenantConnectionProviderImpl.java
index 5a5d715bbd72..e9257cee19c9 100644
--- a/hibernate-core/src/main/java/org/hibernate/engine/jdbc/connections/spi/DataSourceBasedMultiTenantConnectionProviderImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/engine/jdbc/connections/spi/DataSourceBasedMultiTenantConnectionProviderImpl.java
@@ -87,14 +87,14 @@ public void injectServices(ServiceRegistryImplementor serviceRegistry) {
 			throw new HibernateException( "JNDI name [" + this.jndiName + "] could not be resolved" );
 		}
 		else if ( namedObject instanceof DataSource datasource ) {
-			final int loc = this.jndiName.lastIndexOf( '/' );
-			baseJndiNamespace = this.jndiName.substring( 0, loc );
-			final String prefix = this.jndiName.substring( loc + 1);
+			final int loc = jndiName.lastIndexOf( '/' );
+			baseJndiNamespace = jndiName.substring( 0, loc );
+			final String prefix = jndiName.substring( loc + 1);
 			tenantIdentifierForAny = (T) prefix;
 			dataSourceMap().put( tenantIdentifierForAny, datasource );
 		}
 		else if ( namedObject instanceof Context ) {
-			baseJndiNamespace = this.jndiName;
+			baseJndiNamespace = jndiName;
 			final Object configuredTenantId =
 					configurationService.getSettings().get( TENANT_IDENTIFIER_TO_USE_FOR_ANY_KEY );
 			tenantIdentifierForAny = (T) configuredTenantId;
diff --git a/hibernate-core/src/main/java/org/hibernate/engine/jdbc/env/internal/JdbcEnvironmentInitiator.java b/hibernate-core/src/main/java/org/hibernate/engine/jdbc/env/internal/JdbcEnvironmentInitiator.java
index cb57c37c9c0b..c8a4a645df4e 100644
--- a/hibernate-core/src/main/java/org/hibernate/engine/jdbc/env/internal/JdbcEnvironmentInitiator.java
+++ b/hibernate-core/src/main/java/org/hibernate/engine/jdbc/env/internal/JdbcEnvironmentInitiator.java
@@ -806,5 +806,13 @@ public boolean isActive() {
 		public SqlExceptionHelper getSqlExceptionHelper() {
 			return sqlExceptionHelper;
 		}
+
+		@Override
+		public void afterObtainConnection(Connection connection) {
+		}
+
+		@Override
+		public void beforeReleaseConnection(Connection connection) {
+		}
 	}
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/engine/jdbc/internal/JdbcCoordinatorImpl.java b/hibernate-core/src/main/java/org/hibernate/engine/jdbc/internal/JdbcCoordinatorImpl.java
index d962cc0c556a..f4587a1e4605 100644
--- a/hibernate-core/src/main/java/org/hibernate/engine/jdbc/internal/JdbcCoordinatorImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/engine/jdbc/internal/JdbcCoordinatorImpl.java
@@ -76,23 +76,20 @@ public JdbcCoordinatorImpl(
 			Connection userSuppliedConnection,
 			JdbcSessionOwner owner,
 			JdbcServices jdbcServices) {
+		this.owner = owner;
+		this.jdbcServices = jdbcServices;
 		this.isUserSuppliedConnection = userSuppliedConnection != null;
+		this.logicalConnection = createLogicalConnection( userSuppliedConnection, owner );
+	}
 
+	private static LogicalConnectionImplementor createLogicalConnection(
+			Connection userSuppliedConnection,
+			JdbcSessionOwner owner) {
 		final ResourceRegistry resourceRegistry =
 				new ResourceRegistryStandardImpl( owner.getJdbcSessionContext().getEventHandler() );
-		if ( isUserSuppliedConnection ) {
-			this.logicalConnection = new LogicalConnectionProvidedImpl( userSuppliedConnection, resourceRegistry );
-		}
-		else {
-			this.logicalConnection = new LogicalConnectionManagedImpl(
-					owner.getJdbcConnectionAccess(),
-					owner.getJdbcSessionContext(),
-					owner.getSqlExceptionHelper(),
-					resourceRegistry
-			);
-		}
-		this.owner = owner;
-		this.jdbcServices = jdbcServices;
+		return userSuppliedConnection == null
+				? new LogicalConnectionManagedImpl( owner, resourceRegistry )
+				: new LogicalConnectionProvidedImpl( userSuppliedConnection, resourceRegistry );
 	}
 
 	private JdbcCoordinatorImpl(
@@ -405,7 +402,7 @@ public JdbcResourceTransaction getResourceLocalTransaction() {
 	 */
 	@Override
 	public void serialize(ObjectOutputStream oos) throws IOException {
-		if ( ! isReadyForSerialization() ) {
+		if ( !isReadyForSerialization() ) {
 			throw new HibernateException( "Cannot serialize Session while connected" );
 		}
 		oos.writeBoolean( isUserSuppliedConnection );
@@ -423,21 +420,13 @@ public void serialize(ObjectOutputStream oos) throws IOException {
 	 * @throws IOException Trouble accessing the stream
 	 * @throws ClassNotFoundException Trouble reading the stream
 	 */
-	public static JdbcCoordinatorImpl deserialize(
-			ObjectInputStream ois,
-			JdbcSessionOwner owner) throws IOException, ClassNotFoundException {
+	public static JdbcCoordinatorImpl deserialize(ObjectInputStream ois, JdbcSessionOwner owner)
+			throws IOException, ClassNotFoundException {
 		final boolean isUserSuppliedConnection = ois.readBoolean();
-		final LogicalConnectionImplementor logicalConnection;
-		if ( isUserSuppliedConnection ) {
-			logicalConnection = LogicalConnectionProvidedImpl.deserialize( ois );
-		}
-		else {
-			logicalConnection = LogicalConnectionManagedImpl.deserialize(
-					ois,
-					owner.getJdbcConnectionAccess(),
-					owner.getJdbcSessionContext()
-			);
-		}
+		final var logicalConnection =
+				isUserSuppliedConnection
+						? LogicalConnectionProvidedImpl.deserialize( ois )
+						: LogicalConnectionManagedImpl.deserialize( ois, owner );
 		return new JdbcCoordinatorImpl( logicalConnection, isUserSuppliedConnection, owner );
 	}
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/engine/spi/SessionDelegatorBaseImpl.java b/hibernate-core/src/main/java/org/hibernate/engine/spi/SessionDelegatorBaseImpl.java
index 63ad547da9eb..a3a3ce53d2f1 100644
--- a/hibernate-core/src/main/java/org/hibernate/engine/spi/SessionDelegatorBaseImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/engine/spi/SessionDelegatorBaseImpl.java
@@ -67,6 +67,8 @@
 import org.hibernate.stat.SessionStatistics;
 import org.hibernate.type.format.FormatMapper;
 
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
@@ -1247,4 +1249,14 @@ public SessionAssociationMarkers getSessionAssociationMarkers() {
 	public boolean isIdentifierRollbackEnabled() {
 		return delegate.isIdentifierRollbackEnabled();
 	}
+
+	@Override
+	public void afterObtainConnection(Connection connection) throws SQLException {
+		delegate.afterObtainConnection( connection );
+	}
+
+	@Override
+	public void beforeReleaseConnection(Connection connection) throws SQLException {
+		delegate.beforeReleaseConnection( connection );
+	}
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/engine/spi/SharedSessionDelegatorBaseImpl.java b/hibernate-core/src/main/java/org/hibernate/engine/spi/SharedSessionDelegatorBaseImpl.java
index 351216a702eb..2031afbee78e 100644
--- a/hibernate-core/src/main/java/org/hibernate/engine/spi/SharedSessionDelegatorBaseImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/engine/spi/SharedSessionDelegatorBaseImpl.java
@@ -44,6 +44,8 @@
 import org.hibernate.resource.transaction.spi.TransactionCoordinator;
 import org.hibernate.type.format.FormatMapper;
 
+import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.List;
 import java.util.Set;
 import java.util.TimeZone;
@@ -703,4 +705,14 @@ public SessionAssociationMarkers getSessionAssociationMarkers() {
 	public boolean isIdentifierRollbackEnabled() {
 		return delegate.isIdentifierRollbackEnabled();
 	}
+
+	@Override
+	public void afterObtainConnection(Connection connection) throws SQLException {
+		delegate.afterObtainConnection( connection );
+	}
+
+	@Override
+	public void beforeReleaseConnection(Connection connection) throws SQLException {
+		delegate.beforeReleaseConnection(  connection );
+	}
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/generator/internal/TenantIdGeneration.java b/hibernate-core/src/main/java/org/hibernate/generator/internal/TenantIdGeneration.java
index 8efc6cbeb5a8..47ee016b1ba0 100644
--- a/hibernate-core/src/main/java/org/hibernate/generator/internal/TenantIdGeneration.java
+++ b/hibernate-core/src/main/java/org/hibernate/generator/internal/TenantIdGeneration.java
@@ -9,7 +9,6 @@
 
 import org.hibernate.PropertyValueException;
 import org.hibernate.annotations.TenantId;
-import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.generator.BeforeExecutionGenerator;
@@ -51,8 +50,7 @@ public Object generate(SharedSessionContractImplementor session, Object owner, O
 		final SessionFactoryImplementor sessionFactory = session.getSessionFactory();
 		final Object tenantId = session.getTenantIdentifierValue();
 		if ( currentValue != null ) {
-			final CurrentTenantIdentifierResolver<Object> resolver =
-					sessionFactory.getCurrentTenantIdentifierResolver();
+			final var resolver = sessionFactory.getCurrentTenantIdentifierResolver();
 			if ( resolver != null && resolver.isRoot( tenantId ) ) {
 				// the "root" tenant is allowed to set the tenant id explicitly
 				return currentValue;
diff --git a/hibernate-core/src/main/java/org/hibernate/internal/AbstractSharedSessionContract.java b/hibernate-core/src/main/java/org/hibernate/internal/AbstractSharedSessionContract.java
index 1ba70aa4843d..a986bbb1e625 100644
--- a/hibernate-core/src/main/java/org/hibernate/internal/AbstractSharedSessionContract.java
+++ b/hibernate-core/src/main/java/org/hibernate/internal/AbstractSharedSessionContract.java
@@ -28,7 +28,6 @@
 import org.hibernate.boot.spi.SessionFactoryOptions;
 import org.hibernate.bytecode.enhance.spi.interceptor.SessionAssociationMarkers;
 import org.hibernate.cache.spi.CacheTransactionSynchronization;
-import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
 import org.hibernate.dialect.Dialect;
 import org.hibernate.engine.internal.SessionEventListenerManagerImpl;
 import org.hibernate.engine.jdbc.LobCreator;
@@ -108,6 +107,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.io.Serial;
+import java.sql.Connection;
 import java.sql.SQLException;
 import java.util.List;
 import java.util.Locale;
@@ -117,6 +117,7 @@
 import java.util.function.Function;
 
 import static java.lang.Boolean.TRUE;
+import static org.hibernate.boot.model.naming.Identifier.toIdentifier;
 import static org.hibernate.internal.util.StringHelper.isEmpty;
 import static org.hibernate.query.sqm.internal.SqmUtil.verifyIsSelectStatement;
 
@@ -245,7 +246,7 @@ protected final void setUpMultitenancy(SessionFactoryImplementor factory, LoadQu
 				throw new HibernateException( "SessionFactory configured for multi-tenancy, but no tenant identifier specified" );
 			}
 			else {
-				final CurrentTenantIdentifierResolver<Object> resolver = factory.getCurrentTenantIdentifierResolver();
+				final var resolver = factory.getCurrentTenantIdentifierResolver();
 				if ( resolver==null || !resolver.isRoot( tenantIdentifier ) ) {
 					// turn on the filter, unless this is the "root" tenant with access to all partitions
 					loadQueryInfluencers.enableFilter( TenantIdBinder.FILTER_NAME )
@@ -683,9 +684,10 @@ public boolean isConnected() {
 
 	@Override
 	public JdbcConnectionAccess getJdbcConnectionAccess() {
-		// See class-level JavaDocs for a discussion of the concurrent-access safety of this method
+		// See class-level Javadoc for a discussion of the concurrent-access safety of this method
 		if ( jdbcConnectionAccess == null ) {
 			if ( !factoryOptions.isMultiTenancyEnabled() ) {
+				// we might still be using schema-based multitenancy
 				jdbcConnectionAccess = new NonContextualJdbcConnectionAccess(
 						sessionEventsManager,
 						factory.connectionProvider,
@@ -693,6 +695,7 @@ public JdbcConnectionAccess getJdbcConnectionAccess() {
 				);
 			}
 			else {
+				// we're using datasource-based multitenancy
 				jdbcConnectionAccess = new ContextualJdbcConnectionAccess(
 						tenantIdentifier,
 						sessionEventsManager,
@@ -704,6 +707,38 @@ public JdbcConnectionAccess getJdbcConnectionAccess() {
 		return jdbcConnectionAccess;
 	}
 
+	private boolean useSchemaBasedMultiTenancy() {
+		return tenantIdentifier != null
+			&& factory.getSessionFactoryOptions().getTenantSchemaMapper() != null;
+	}
+
+	private String tenantSchema() {
+		final var mapper = factory.getSessionFactoryOptions().getTenantSchemaMapper();
+		return mapper == null ? null : normalizeSchemaName( mapper.schemaName( tenantIdentifier ) );
+	}
+
+	private String normalizeSchemaName(String schemaName) {
+		return jdbcServices.getJdbcEnvironment().getIdentifierHelper()
+				.toMetaDataSchemaName( toIdentifier( schemaName ) );
+	}
+
+	private transient String initialSchema;
+
+	@Override
+	public void afterObtainConnection(Connection connection) throws SQLException {
+		if ( useSchemaBasedMultiTenancy() ) {
+			initialSchema = connection.getSchema();
+			connection.setSchema( tenantSchema() );
+		}
+	}
+
+	@Override
+	public void beforeReleaseConnection(Connection connection) throws SQLException {
+		if ( useSchemaBasedMultiTenancy() ) {
+			connection.setSchema( initialSchema );
+		}
+	}
+
 	@Override
 	public EntityKey generateEntityKey(Object id, EntityPersister persister) {
 		return new EntityKey( id, persister );
diff --git a/hibernate-core/src/main/java/org/hibernate/internal/ContextualJdbcConnectionAccess.java b/hibernate-core/src/main/java/org/hibernate/internal/ContextualJdbcConnectionAccess.java
index 53aead819832..003c99ed45b8 100644
--- a/hibernate-core/src/main/java/org/hibernate/internal/ContextualJdbcConnectionAccess.java
+++ b/hibernate-core/src/main/java/org/hibernate/internal/ContextualJdbcConnectionAccess.java
@@ -44,17 +44,13 @@ public Connection obtainConnection() throws SQLException {
 		}
 
 		final EventMonitor eventMonitor = session.getEventMonitor();
-		final DiagnosticEvent jdbcConnectionAcquisitionEvent = eventMonitor.beginJdbcConnectionAcquisitionEvent();
+		final DiagnosticEvent connectionAcquisitionEvent = eventMonitor.beginJdbcConnectionAcquisitionEvent();
 		try {
 			listener.jdbcConnectionAcquisitionStart();
 			return connectionProvider.getConnection( tenantIdentifier );
 		}
 		finally {
-			eventMonitor.completeJdbcConnectionAcquisitionEvent(
-					jdbcConnectionAcquisitionEvent,
-					session,
-					tenantIdentifier
-			);
+			eventMonitor.completeJdbcConnectionAcquisitionEvent( connectionAcquisitionEvent, session, tenantIdentifier );
 			listener.jdbcConnectionAcquisitionEnd();
 		}
 	}
@@ -66,13 +62,13 @@ public void releaseConnection(Connection connection) throws SQLException {
 		}
 
 		final EventMonitor eventMonitor = session.getEventMonitor();
-		final DiagnosticEvent jdbcConnectionReleaseEvent = eventMonitor.beginJdbcConnectionReleaseEvent();
+		final DiagnosticEvent connectionReleaseEvent = eventMonitor.beginJdbcConnectionReleaseEvent();
 		try {
 			listener.jdbcConnectionReleaseStart();
 			connectionProvider.releaseConnection( tenantIdentifier, connection );
 		}
 		finally {
-			eventMonitor.completeJdbcConnectionReleaseEvent( jdbcConnectionReleaseEvent, session, tenantIdentifier );
+			eventMonitor.completeJdbcConnectionReleaseEvent( connectionReleaseEvent, session, tenantIdentifier );
 			listener.jdbcConnectionReleaseEnd();
 		}
 	}
diff --git a/hibernate-core/src/main/java/org/hibernate/internal/NonContextualJdbcConnectionAccess.java b/hibernate-core/src/main/java/org/hibernate/internal/NonContextualJdbcConnectionAccess.java
index 8f91d9e67930..96ae39e771e3 100644
--- a/hibernate-core/src/main/java/org/hibernate/internal/NonContextualJdbcConnectionAccess.java
+++ b/hibernate-core/src/main/java/org/hibernate/internal/NonContextualJdbcConnectionAccess.java
@@ -38,17 +38,13 @@ public NonContextualJdbcConnectionAccess(
 	@Override
 	public Connection obtainConnection() throws SQLException {
 		final EventMonitor eventMonitor = session.getEventMonitor();
-		final DiagnosticEvent jdbcConnectionAcquisitionEvent = eventMonitor.beginJdbcConnectionAcquisitionEvent();
+		final DiagnosticEvent connectionAcquisitionEvent = eventMonitor.beginJdbcConnectionAcquisitionEvent();
 		try {
 			listener.jdbcConnectionAcquisitionStart();
 			return connectionProvider.getConnection();
 		}
 		finally {
-			eventMonitor.completeJdbcConnectionAcquisitionEvent(
-					jdbcConnectionAcquisitionEvent,
-					session,
-					null
-			);
+			eventMonitor.completeJdbcConnectionAcquisitionEvent( connectionAcquisitionEvent, session, null );
 			listener.jdbcConnectionAcquisitionEnd();
 		}
 	}
@@ -56,13 +52,13 @@ public Connection obtainConnection() throws SQLException {
 	@Override
 	public void releaseConnection(Connection connection) throws SQLException {
 		final EventMonitor eventMonitor = session.getEventMonitor();
-		final DiagnosticEvent jdbcConnectionReleaseEvent = eventMonitor.beginJdbcConnectionReleaseEvent();
+		final DiagnosticEvent connectionReleaseEvent = eventMonitor.beginJdbcConnectionReleaseEvent();
 		try {
 			listener.jdbcConnectionReleaseStart();
 			connectionProvider.closeConnection( connection );
 		}
 		finally {
-			eventMonitor.completeJdbcConnectionReleaseEvent( jdbcConnectionReleaseEvent, session, null );
+			eventMonitor.completeJdbcConnectionReleaseEvent( connectionReleaseEvent, session, null );
 			listener.jdbcConnectionReleaseEnd();
 		}
 	}
diff --git a/hibernate-core/src/main/java/org/hibernate/internal/SessionFactoryImpl.java b/hibernate-core/src/main/java/org/hibernate/internal/SessionFactoryImpl.java
index f8200fd0b5d8..86e6ec48ff31 100644
--- a/hibernate-core/src/main/java/org/hibernate/internal/SessionFactoryImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/internal/SessionFactoryImpl.java
@@ -458,7 +458,7 @@ private void disintegrate(Exception startupException, IntegratorObserver integra
 
 
 	private SessionBuilderImpl createDefaultSessionOpenOptionsIfPossible() {
-		final CurrentTenantIdentifierResolver<Object> tenantIdResolver = getCurrentTenantIdentifierResolver();
+		final var tenantIdResolver = getCurrentTenantIdentifierResolver();
 		if ( tenantIdResolver == null ) {
 			return withOptions();
 		}
@@ -1154,7 +1154,7 @@ public SessionBuilderImpl(SessionFactoryImpl sessionFactory) {
 			subselectFetchEnabled = sessionFactoryOptions.isSubselectFetchEnabled();
 			identifierRollback = sessionFactoryOptions.isIdentifierRollbackEnabled();
 
-			final CurrentTenantIdentifierResolver<Object> currentTenantIdentifierResolver =
+			final var currentTenantIdentifierResolver =
 					sessionFactory.getCurrentTenantIdentifierResolver();
 			if ( currentTenantIdentifierResolver != null ) {
 				tenantIdentifier = currentTenantIdentifierResolver.resolveCurrentTenantIdentifier();
diff --git a/hibernate-core/src/main/java/org/hibernate/relational/SchemaManager.java b/hibernate-core/src/main/java/org/hibernate/relational/SchemaManager.java
index 9b0ba453a69e..55cd0f40c28f 100644
--- a/hibernate-core/src/main/java/org/hibernate/relational/SchemaManager.java
+++ b/hibernate-core/src/main/java/org/hibernate/relational/SchemaManager.java
@@ -19,7 +19,7 @@
  * {@link jakarta.persistence.SchemaManager}, which it now inherits,
  * with a minor change to the naming of its operations. It is retained
  * for backward compatibility and as a place to define additional
- * functionality such as {@link #populate()}.
+ * functionality such as {@link #populate} and {@link #forSchema}.
  *
  * @since 6.2
  * @author Gavin King
@@ -94,4 +94,28 @@ public interface SchemaManager extends jakarta.persistence.SchemaManager {
 	 */
 	@Incubating
 	void populate();
+
+	/**
+	 * Obtain an instance which targets the given schema.
+	 * <p>
+	 * This is especially useful when using multiple schemas, for example, in
+	 * {@linkplain org.hibernate.cfg.MultiTenancySettings#MULTI_TENANT_SCHEMA_MAPPER
+	 * schema-based multitenancy}.
+	 *
+	 * @param schemaName The name of the schema to target
+	 *
+	 * @since 7.1
+	 */
+	@Incubating
+	SchemaManager forSchema(String schemaName);
+
+	/**
+	 * Obtain an instance which targets the given catalog.
+	 *
+	 * @param catalogName The name of the catalog to target
+	 *
+	 * @since 7.1
+	 */
+	@Incubating
+	SchemaManager forCatalog(String catalogName);
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/relational/internal/SchemaManagerImpl.java b/hibernate-core/src/main/java/org/hibernate/relational/internal/SchemaManagerImpl.java
index c8b2fbae9c86..20f39ae51056 100644
--- a/hibernate-core/src/main/java/org/hibernate/relational/internal/SchemaManagerImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/relational/internal/SchemaManagerImpl.java
@@ -26,12 +26,42 @@
 public class SchemaManagerImpl implements SchemaManager {
 	private final SessionFactoryImplementor sessionFactory;
 	private final MetadataImplementor metadata;
+	private final String schemaName;
+	private final String catalogName;
 
 	public SchemaManagerImpl(
 			SessionFactoryImplementor sessionFactory,
 			MetadataImplementor metadata) {
+		this( sessionFactory, metadata, null, null );
+	}
+
+	public SchemaManagerImpl(
+			SessionFactoryImplementor sessionFactory,
+			MetadataImplementor metadata,
+			String schemaName, String catalogName) {
 		this.sessionFactory = sessionFactory;
 		this.metadata = metadata;
+		this.schemaName = schemaName;
+		this.catalogName = catalogName;
+	}
+
+	@Override
+	public SchemaManager forSchema(String schemaName) {
+		return new SchemaManagerImpl( sessionFactory, metadata, schemaName, catalogName );
+	}
+
+	@Override
+	public SchemaManager forCatalog(String catalogName) {
+		return new SchemaManagerImpl( sessionFactory, metadata, schemaName, catalogName );
+	}
+
+	private void addSchemaAndCatalog(Map<String, Object> properties) {
+		if ( schemaName != null ) {
+			properties.put( AvailableSettings.DEFAULT_SCHEMA, schemaName );
+		}
+		if ( catalogName != null ) {
+			properties.put( AvailableSettings.DEFAULT_CATALOG, catalogName );
+		}
 	}
 
 	@Override
@@ -40,6 +70,7 @@ public void exportMappedObjects(boolean createSchemas) {
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_DATABASE_ACTION, Action.CREATE_ONLY );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_SCRIPTS_ACTION, Action.NONE );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_CREATE_SCHEMAS, createSchemas );
+		addSchemaAndCatalog( properties );
 		SchemaManagementToolCoordinator.process(
 				metadata,
 				sessionFactory.getServiceRegistry(),
@@ -54,6 +85,7 @@ public void dropMappedObjects(boolean dropSchemas) {
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_DATABASE_ACTION, Action.DROP );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_SCRIPTS_ACTION, Action.NONE );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_CREATE_SCHEMAS, dropSchemas );
+		addSchemaAndCatalog( properties );
 		SchemaManagementToolCoordinator.process(
 				metadata,
 				sessionFactory.getServiceRegistry(),
@@ -68,6 +100,7 @@ public void validateMappedObjects() {
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_DATABASE_ACTION, Action.VALIDATE );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_SCRIPTS_ACTION, Action.NONE );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_CREATE_SCHEMAS, false );
+		addSchemaAndCatalog( properties );
 		SchemaManagementToolCoordinator.process(
 				metadata,
 				sessionFactory.getServiceRegistry(),
@@ -81,6 +114,7 @@ public void truncateMappedObjects() {
 		Map<String, Object> properties = new HashMap<>( sessionFactory.getProperties() );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_DATABASE_ACTION, Action.TRUNCATE );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_SCRIPTS_ACTION, Action.NONE );
+		addSchemaAndCatalog( properties );
 		SchemaManagementToolCoordinator.process(
 				metadata,
 				sessionFactory.getServiceRegistry(),
@@ -94,6 +128,7 @@ public void populate() {
 		Map<String, Object> properties = new HashMap<>( sessionFactory.getProperties() );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_DATABASE_ACTION, Action.POPULATE );
 		properties.put( AvailableSettings.JAKARTA_HBM2DDL_SCRIPTS_ACTION, Action.NONE );
+		addSchemaAndCatalog( properties );
 		SchemaManagementToolCoordinator.process(
 				metadata,
 				sessionFactory.getServiceRegistry(),
diff --git a/hibernate-core/src/main/java/org/hibernate/resource/jdbc/internal/LogicalConnectionManagedImpl.java b/hibernate-core/src/main/java/org/hibernate/resource/jdbc/internal/LogicalConnectionManagedImpl.java
index 84b496294af9..0ef4cfd056e4 100644
--- a/hibernate-core/src/main/java/org/hibernate/resource/jdbc/internal/LogicalConnectionManagedImpl.java
+++ b/hibernate-core/src/main/java/org/hibernate/resource/jdbc/internal/LogicalConnectionManagedImpl.java
@@ -12,14 +12,13 @@
 
 import org.hibernate.ResourceClosedException;
 import org.hibernate.engine.jdbc.connections.spi.JdbcConnectionAccess;
-import org.hibernate.engine.jdbc.spi.JdbcServices;
 import org.hibernate.engine.jdbc.spi.SqlExceptionHelper;
 import org.hibernate.internal.CoreLogging;
 import org.hibernate.internal.CoreMessageLogger;
 import org.hibernate.resource.jdbc.LogicalConnection;
 import org.hibernate.resource.jdbc.ResourceRegistry;
-import org.hibernate.resource.jdbc.spi.JdbcEventHandler;
 import org.hibernate.resource.jdbc.spi.JdbcSessionContext;
+import org.hibernate.resource.jdbc.spi.JdbcSessionOwner;
 import org.hibernate.resource.jdbc.spi.PhysicalConnectionHandlingMode;
 
 import static org.hibernate.ConnectionAcquisitionMode.IMMEDIATELY;
@@ -39,94 +38,75 @@
 public class LogicalConnectionManagedImpl extends AbstractLogicalConnectionImplementor {
 	private static final CoreMessageLogger log = CoreLogging.messageLogger( LogicalConnectionManagedImpl.class );
 
-	private final transient JdbcConnectionAccess jdbcConnectionAccess;
-	private final transient JdbcEventHandler jdbcEventHandler;
-	private final transient SqlExceptionHelper sqlExceptionHelper;
-
+	private final transient JdbcSessionOwner jdbcSessionOwner;
 	private final transient PhysicalConnectionHandlingMode connectionHandlingMode;
 
 	private transient Connection physicalConnection;
 	private boolean closed;
 
-	private final boolean providerDisablesAutoCommit;
-
-	public LogicalConnectionManagedImpl(
-			JdbcConnectionAccess jdbcConnectionAccess,
-			JdbcSessionContext jdbcSessionContext,
-			SqlExceptionHelper sqlExceptionHelper,
-			ResourceRegistry resourceRegistry) {
-		this.jdbcConnectionAccess = jdbcConnectionAccess;
-		this.jdbcEventHandler = jdbcSessionContext.getEventHandler();
+	public LogicalConnectionManagedImpl(JdbcSessionOwner sessionOwner, ResourceRegistry resourceRegistry) {
+		this.jdbcSessionOwner = sessionOwner;
 		this.resourceRegistry = resourceRegistry;
 
-		this.connectionHandlingMode = determineConnectionHandlingMode(
-				jdbcSessionContext.getPhysicalConnectionHandlingMode(),
-				jdbcConnectionAccess );
-
-		this.sqlExceptionHelper = sqlExceptionHelper;
-
+		connectionHandlingMode = determineConnectionHandlingMode( sessionOwner );
 		if ( connectionHandlingMode.getAcquisitionMode() == IMMEDIATELY ) {
 			//noinspection resource
 			acquireConnectionIfNeeded();
 		}
 
-		this.providerDisablesAutoCommit = jdbcSessionContext.doesConnectionProviderDisableAutoCommit();
-		if ( providerDisablesAutoCommit ) {
+		if ( sessionOwner.getJdbcSessionContext().doesConnectionProviderDisableAutoCommit() ) {
 			log.connectionProviderDisablesAutoCommitEnabled();
 		}
 	}
 
-	public LogicalConnectionManagedImpl(
-			JdbcConnectionAccess jdbcConnectionAccess,
-			JdbcSessionContext jdbcSessionContext,
-			ResourceRegistry resourceRegistry,
-			JdbcServices jdbcServices) {
-		this(
-				jdbcConnectionAccess,
-				jdbcSessionContext,
-				jdbcServices.getSqlExceptionHelper(),
-				resourceRegistry
-		);
-	}
-
-	private PhysicalConnectionHandlingMode determineConnectionHandlingMode(
-			PhysicalConnectionHandlingMode connectionHandlingMode,
-			JdbcConnectionAccess jdbcConnectionAccess) {
+	private PhysicalConnectionHandlingMode determineConnectionHandlingMode(JdbcSessionOwner sessionOwner) {
+		final var connectionHandlingMode = sessionOwner.getJdbcSessionContext().getPhysicalConnectionHandlingMode();
 		return connectionHandlingMode.getReleaseMode() == AFTER_STATEMENT
-			&& !jdbcConnectionAccess.supportsAggressiveRelease()
+			&& !sessionOwner.getJdbcConnectionAccess().supportsAggressiveRelease()
 				? DELAYED_ACQUISITION_AND_RELEASE_AFTER_TRANSACTION
 				: connectionHandlingMode;
 	}
 
-	private LogicalConnectionManagedImpl(
-			JdbcConnectionAccess jdbcConnectionAccess,
-			JdbcSessionContext jdbcSessionContext,
-			boolean closed) {
-		this(
-				jdbcConnectionAccess,
-				jdbcSessionContext,
-				new ResourceRegistryStandardImpl(),
-				jdbcSessionContext.getJdbcServices()
-		);
+	private LogicalConnectionManagedImpl(JdbcSessionOwner owner, boolean closed) {
+		this( owner, new ResourceRegistryStandardImpl() );
 		this.closed = closed;
 	}
 
+	private JdbcSessionContext getJdbcSessionContext() {
+		return jdbcSessionOwner.getJdbcSessionContext();
+	}
+
+	private JdbcConnectionAccess getJdbcConnectionAccess() {
+		return jdbcSessionOwner.getJdbcConnectionAccess();
+	}
+
+	private SqlExceptionHelper getExceptionHelper() {
+		return jdbcSessionOwner.getSqlExceptionHelper();
+	}
+
 	private Connection acquireConnectionIfNeeded() {
 		if ( physicalConnection == null ) {
-			jdbcEventHandler.jdbcConnectionAcquisitionStart();
-			try {
-				physicalConnection = jdbcConnectionAccess.obtainConnection();
-			}
-			catch ( SQLException e ) {
-				throw sqlExceptionHelper.convert( e, "Unable to acquire JDBC Connection" );
-			}
-			finally {
-				jdbcEventHandler.jdbcConnectionAcquisitionEnd( physicalConnection );
-			}
+			physicalConnection = acquire();
+			afterAcquire();
 		}
 		return physicalConnection;
 	}
 
+	private void releaseConnectionIfNeeded() {
+		final Connection connection = physicalConnection;
+		if ( connection != null ) {
+			beforeRelease();
+			// Set the connection to null before releasing resources to prevent
+			// recursion into this method. Recursion can happen when we release
+			// resources and when batch statements are in progress: releasing
+			// resources aborts the batch statement, which then triggers
+			// logicalConnection.afterStatement(), which in some configurations
+			// releases the connection.
+			physicalConnection = null;
+			release( connection );
+		}
+	}
+
 	@Override
 	public boolean isOpen() {
 		return !closed;
@@ -157,7 +137,7 @@ public void afterStatement() {
 			}
 			else {
 				log.debug( "Initiating JDBC connection release from afterStatement" );
-				releaseConnection();
+				releaseConnectionIfNeeded();
 			}
 		}
 	}
@@ -167,7 +147,7 @@ public void beforeTransactionCompletion() {
 		super.beforeTransactionCompletion();
 		if ( connectionHandlingMode.getReleaseMode() == BEFORE_TRANSACTION_COMPLETION ) {
 			log.debug( "Initiating JDBC connection release from beforeTransactionCompletion" );
-			releaseConnection();
+			releaseConnectionIfNeeded();
 		}
 	}
 
@@ -175,12 +155,12 @@ public void beforeTransactionCompletion() {
 	public void afterTransaction() {
 		super.afterTransaction();
 		if ( connectionHandlingMode.getReleaseMode() != ON_CLOSE ) {
-			// NOTE : we check for !ON_CLOSE here (rather than AFTER_TRANSACTION) to also catch:
+			// NOTE: we check for !ON_CLOSE here (rather than AFTER_TRANSACTION) to also catch:
 			// - AFTER_STATEMENT cases that were circumvented due to held resources
 			// - BEFORE_TRANSACTION_COMPLETION cases that were circumvented because a rollback occurred
 			//   (we don't get a beforeTransactionCompletion event on rollback).
 			log.debug( "Initiating JDBC connection release from afterTransaction" );
-			releaseConnection();
+			releaseConnectionIfNeeded();
 		}
 	}
 
@@ -190,7 +170,7 @@ public Connection manualDisconnect() {
 			throw new ResourceClosedException( "Logical connection is closed" );
 		}
 		final Connection connection = physicalConnection;
-		releaseConnection();
+		releaseConnectionIfNeeded();
 		return connection;
 	}
 
@@ -202,35 +182,66 @@ public void manualReconnect(Connection suppliedConnection) {
 		throw new IllegalStateException( "Cannot manually reconnect unless Connection was originally supplied by user" );
 	}
 
-	private void releaseConnection() {
-		final Connection localVariableConnection = physicalConnection;
-		if ( localVariableConnection != null ) {
-			// We need to set the connection to null before we release resources,
-			// in order to prevent recursion into this method.
-			// Recursion can happen when we release resources and when batch statements are in progress:
-			// when releasing resources, we'll abort the batch statement,
-			// which will trigger "logicalConnection.afterStatement()",
-			// which in some configurations will release the connection.
-			physicalConnection = null;
+	private Connection acquire() {
+		final var eventHandler = getJdbcSessionContext().getEventHandler();
+		eventHandler.jdbcConnectionAcquisitionStart();
+		try {
+			return getJdbcConnectionAccess().obtainConnection();
+		}
+		catch ( SQLException e ) {
+			throw getExceptionHelper().convert( e, "Unable to acquire JDBC Connection" );
+		}
+		finally {
+			eventHandler.jdbcConnectionAcquisitionEnd( physicalConnection );
+		}
+	}
+
+	private void release(Connection connection) {
+		final var eventHandler = getJdbcSessionContext().getEventHandler();
+		try {
 			try {
-				try {
-					getResourceRegistry().releaseResources();
-					if ( !localVariableConnection.isClosed() ) {
-						sqlExceptionHelper.logAndClearWarnings( localVariableConnection );
-					}
+				getResourceRegistry().releaseResources();
+				if ( !connection.isClosed() ) {
+					getExceptionHelper().logAndClearWarnings( connection );
 				}
-				finally {
-					jdbcEventHandler.jdbcConnectionReleaseStart();
-					jdbcConnectionAccess.releaseConnection( localVariableConnection );
-				}
-			}
-			catch (SQLException e) {
-				throw sqlExceptionHelper.convert( e, "Unable to release JDBC Connection" );
 			}
 			finally {
-				jdbcEventHandler.jdbcConnectionReleaseEnd();
+				eventHandler.jdbcConnectionReleaseStart();
+				getJdbcConnectionAccess().releaseConnection( connection );
 			}
 		}
+		catch (SQLException e) {
+			throw getExceptionHelper().convert( e, "Unable to release JDBC Connection" );
+		}
+		finally {
+			eventHandler.jdbcConnectionReleaseEnd();
+		}
+	}
+
+	private void afterAcquire() {
+		try {
+			// give the session a chance to set the schema
+			jdbcSessionOwner.afterObtainConnection( physicalConnection );
+		}
+		catch (SQLException e) {
+			try {
+				getJdbcConnectionAccess().releaseConnection( physicalConnection );
+			}
+			catch (SQLException re) {
+				e.addSuppressed( re );
+			}
+			throw getExceptionHelper().convert( e, "Error after acquiring JDBC Connection" );
+		}
+	}
+
+	private void beforeRelease() {
+		try {
+			// give the session a chance to change the schema back to null
+			jdbcSessionOwner.beforeReleaseConnection( physicalConnection );
+		}
+		catch (SQLException e) {
+			log.warn( "Error before releasing JDBC connection", e );
+		}
 	}
 
 	@Override
@@ -238,12 +249,9 @@ public void serialize(ObjectOutputStream oos) throws IOException {
 		oos.writeBoolean( closed );
 	}
 
-	public static LogicalConnectionManagedImpl deserialize(
-			ObjectInputStream ois,
-			JdbcConnectionAccess jdbcConnectionAccess,
-			JdbcSessionContext jdbcSessionContext) throws IOException {
-		final boolean isClosed = ois.readBoolean();
-		return new LogicalConnectionManagedImpl( jdbcConnectionAccess, jdbcSessionContext, isClosed );
+	public static LogicalConnectionManagedImpl deserialize(ObjectInputStream ois, JdbcSessionOwner owner)
+			throws IOException {
+		return new LogicalConnectionManagedImpl( owner, ois.readBoolean() );
 	}
 
 	@Override
@@ -252,7 +260,7 @@ public Connection close() {
 			getResourceRegistry().releaseResources();
 			log.closingLogicalConnection();
 			try {
-				releaseConnection();
+				releaseConnectionIfNeeded();
 			}
 			finally {
 				// no matter what
@@ -275,8 +283,9 @@ protected Connection getConnectionForTransactionManagement() {
 
 	@Override
 	public void begin() {
-		initiallyAutoCommit = !doConnectionsFromProviderHaveAutoCommitDisabled()
-				&& determineInitialAutoCommitMode( getConnectionForTransactionManagement() );
+		initiallyAutoCommit =
+				!doConnectionsFromProviderHaveAutoCommitDisabled()
+						&& determineInitialAutoCommitMode( getConnectionForTransactionManagement() );
 		super.begin();
 	}
 
@@ -289,6 +298,6 @@ protected void afterCompletion() {
 
 	@Override
 	protected boolean doConnectionsFromProviderHaveAutoCommitDisabled() {
-		return providerDisablesAutoCommit;
+		return getJdbcSessionContext().doesConnectionProviderDisableAutoCommit();
 	}
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/resource/jdbc/spi/JdbcSessionOwner.java b/hibernate-core/src/main/java/org/hibernate/resource/jdbc/spi/JdbcSessionOwner.java
index 4f47a673a526..310433ef825f 100644
--- a/hibernate-core/src/main/java/org/hibernate/resource/jdbc/spi/JdbcSessionOwner.java
+++ b/hibernate-core/src/main/java/org/hibernate/resource/jdbc/spi/JdbcSessionOwner.java
@@ -4,12 +4,16 @@
  */
 package org.hibernate.resource.jdbc.spi;
 
+import org.hibernate.Incubating;
 import org.hibernate.engine.jdbc.connections.spi.JdbcConnectionAccess;
 import org.hibernate.engine.jdbc.spi.SqlExceptionHelper;
 import org.hibernate.event.spi.EventManager;
 import org.hibernate.event.monitor.spi.EventMonitor;
 import org.hibernate.resource.transaction.spi.TransactionCoordinator;
 
+import java.sql.Connection;
+import java.sql.SQLException;
+
 import static java.lang.reflect.InvocationHandler.invokeDefault;
 import static java.lang.reflect.Proxy.newProxyInstance;
 
@@ -78,6 +82,35 @@ default SqlExceptionHelper getSqlExceptionHelper() {
 		return getJdbcSessionContext().getJdbcServices().getSqlExceptionHelper();
 	}
 
+	/**
+	 * Called after a managed JDBC connection is obtained.
+	 * <p>
+	 * Sets the schema to the schema belonging to the current tenant if:
+	 * <ol>
+	 * <li>{@value org.hibernate.cfg.MultiTenancySettings#MULTI_TENANT_SCHEMA_MAPPER} is enabled, and
+	 * <li>this session has an active tenant id.
+	 * </ol>
+	 *
+	 * @param connection A JDBC connection which was just acquired
+	 *
+	 * @since 7.1
+	 */
+	@Incubating
+	void afterObtainConnection(Connection connection) throws SQLException;
+
+	/**
+	 * Called right before a managed JDBC connection is released.
+	 * <p>
+	 * Unset the schema which was set by {@link #afterObtainConnection},
+	 * if any.
+	 * .
+	 * @param connection The JDBC connection which is about to be released
+	 *
+	 * @since 7.1
+	 */
+	@Incubating
+	void beforeReleaseConnection(Connection connection) throws SQLException;
+
 	/**
 	 * Obtain a reference to the {@link EventMonitor}.
 	 *
diff --git a/hibernate-core/src/main/java/org/hibernate/tool/schema/spi/SchemaManagementToolCoordinator.java b/hibernate-core/src/main/java/org/hibernate/tool/schema/spi/SchemaManagementToolCoordinator.java
index b3b4ceaedbc3..c415099d5188 100644
--- a/hibernate-core/src/main/java/org/hibernate/tool/schema/spi/SchemaManagementToolCoordinator.java
+++ b/hibernate-core/src/main/java/org/hibernate/tool/schema/spi/SchemaManagementToolCoordinator.java
@@ -120,12 +120,12 @@ public static void process(
 				StandardConverters.BOOLEAN,
 				false
 		);
-		final ExceptionHandler exceptionHandler = haltOnError ? ExceptionHandlerHaltImpl.INSTANCE : ExceptionHandlerLoggedImpl.INSTANCE;
 
-		final ExecutionOptions executionOptions = buildExecutionOptions(
-				configurationValues,
-				exceptionHandler
-		);
+		final ExecutionOptions executionOptions =
+				buildExecutionOptions( configurationValues,
+						haltOnError
+								? ExceptionHandlerHaltImpl.INSTANCE
+								: ExceptionHandlerLoggedImpl.INSTANCE );
 
 		if ( scriptActionMap != null ) {
 			scriptActionMap.forEach(
diff --git a/hibernate-core/src/test/java/org/hibernate/orm/test/multitenancy/SchemaBasedMultitenancyTest.java b/hibernate-core/src/test/java/org/hibernate/orm/test/multitenancy/SchemaBasedMultitenancyTest.java
new file mode 100644
index 000000000000..df7d0b72faad
--- /dev/null
+++ b/hibernate-core/src/test/java/org/hibernate/orm/test/multitenancy/SchemaBasedMultitenancyTest.java
@@ -0,0 +1,94 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright Red Hat Inc. and Hibernate Authors
+ */
+package org.hibernate.orm.test.multitenancy;
+
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.hibernate.context.spi.CurrentTenantIdentifierResolver;
+import org.hibernate.context.spi.TenantSchemaMapper;
+import org.hibernate.dialect.SQLServerDialect;
+import org.hibernate.dialect.SybaseASEDialect;
+import org.hibernate.relational.SchemaManager;
+import org.hibernate.testing.orm.junit.DialectFeatureChecks;
+import org.hibernate.testing.orm.junit.EntityManagerFactoryScope;
+import org.hibernate.testing.orm.junit.JiraKey;
+import org.hibernate.testing.orm.junit.Jpa;
+import org.hibernate.testing.orm.junit.RequiresDialectFeature;
+import org.hibernate.testing.orm.junit.Setting;
+import org.hibernate.testing.orm.junit.SkipForDialect;
+import org.junit.jupiter.api.Test;
+
+import static org.hibernate.cfg.MultiTenancySettings.MULTI_TENANT_IDENTIFIER_RESOLVER;
+import static org.hibernate.cfg.MultiTenancySettings.MULTI_TENANT_SCHEMA_MAPPER;
+import static org.junit.jupiter.api.AssertionsKt.assertNotNull;
+import static org.junit.jupiter.api.AssertionsKt.assertNull;
+
+@Jpa(annotatedClasses = {SchemaBasedMultitenancyTest.Person.class},
+		integrationSettings =
+				{@Setting(name = MULTI_TENANT_SCHEMA_MAPPER,
+						value = "org.hibernate.orm.test.multitenancy.SchemaBasedMultitenancyTest$MyMapper"),
+				@Setting(name = MULTI_TENANT_IDENTIFIER_RESOLVER,
+						value = "org.hibernate.orm.test.multitenancy.SchemaBasedMultitenancyTest$MyResolver")})
+@RequiresDialectFeature(feature = DialectFeatureChecks.SupportSchemaCreation.class)
+@SkipForDialect(dialectClass = SQLServerDialect.class, reason = "Warning: setSchema is a no-op in this driver version")
+@SkipForDialect(dialectClass = SybaseASEDialect.class, reason = "getSchema() method not implemented by jTDS")
+@JiraKey("HHH-19559")
+public class SchemaBasedMultitenancyTest {
+	private static String currentTenantIdentifier;
+
+	@Test void test(EntityManagerFactoryScope scope) {
+		var schemaManager = (SchemaManager) scope.getEntityManagerFactory().getSchemaManager();
+		createSchema( schemaManager, "HELLO" );
+		createSchema( schemaManager, "GOODBYE" );
+		currentTenantIdentifier = "hello";
+		scope.inTransaction( session -> {
+			Person person = new Person();
+			person.ssn = "123456789";
+			person.name = "Gavin";
+			session.persist( person );
+		} );
+		scope.inTransaction( session -> {
+			assertNotNull( session.find( Person.class, "123456789" ) );
+		} );
+		currentTenantIdentifier = "goodbye";
+		scope.inTransaction( session -> {
+			assertNull( session.find( Person.class, "123456789" ) );
+		} );
+	}
+
+	private static void createSchema(SchemaManager schemaManager, String schemaName) {
+		SchemaManager managerForTenantSchema = schemaManager.forSchema( schemaName );
+		managerForTenantSchema.drop(true);
+		managerForTenantSchema.create( true );
+	}
+
+	@Entity(name = "PersonForTenant")
+	static class Person {
+		@Id
+		String ssn;
+		private String name;
+	}
+
+	public static class MyResolver implements CurrentTenantIdentifierResolver<String> {
+		@Override
+		public @NonNull String resolveCurrentTenantIdentifier() {
+			return currentTenantIdentifier;
+		}
+
+		@Override
+		public boolean validateExistingCurrentSessions() {
+			return false;
+		}
+	}
+
+	public static class MyMapper implements TenantSchemaMapper<String> {
+		@Override
+		public @NonNull String schemaName(@NonNull String tenantIdentifier) {
+			return tenantIdentifier;
+		}
+	}
+
+}
diff --git a/hibernate-jfr/src/main/java/org/hibernate/event/jfr/internal/JfrEventMonitor.java b/hibernate-jfr/src/main/java/org/hibernate/event/jfr/internal/JfrEventMonitor.java
index 5775c1e7f844..a2c9af2b9111 100644
--- a/hibernate-jfr/src/main/java/org/hibernate/event/jfr/internal/JfrEventMonitor.java
+++ b/hibernate-jfr/src/main/java/org/hibernate/event/jfr/internal/JfrEventMonitor.java
@@ -780,10 +780,7 @@ public void completeCollectionRemoveEvent(
 	}
 
 	private String getSessionIdentifier(SharedSessionContractImplementor session) {
-		if ( session == null ) {
-			return null;
-		}
-		return session.getSessionIdentifier().toString();
+		return session == null ? null : session.getSessionIdentifier().toString();
 	}
 
 	private String getEntityName(EntityPersister persister) {