HHH-13763 : Update all load-by-key handling to use SQL AST

- Preliminary work on NaturalIdLoader

hibernate-core/src/main/java/org/hibernate/internal/util/StringHelper.java

@@ -6,7 +6,6 @@
  */
 package org.hibernate.internal.util;
 
-import java.io.Serializable;
 import java.util.Arrays;
 import java.util.BitSet;
 import java.util.Collection;
@@ -56,6 +55,23 @@ public static String join(String seperator, String[] strings) {
 		return buf.toString();
 	}
 
+	public static String join(String separator, Object[] values) {
+		int length = values.length;
+		if ( length == 0 ) {
+			return "";
+		}
+		// Allocate space for length * firstStringLength;
+		// If strings[0] is null, then its length is defined as 4, since that's the
+		// length of "null".
+		final int firstStringLength = values[0] != null ? values[0].toString().length() : 4;
+		StringBuilder buf = new StringBuilder( length * firstStringLength )
+				.append( values[0] );
+		for ( int i = 1; i < length; i++ ) {
+			buf.append( separator ).append( values[i] );
+		}
+		return buf.toString();
+	}
+
 	public static String join(String separator, Iterable objects) {
 		return join( separator, objects.iterator() );
 	}

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

@@ -39,6 +39,8 @@
 import org.jboss.logging.Logger;
 
 /**
+ * A one-time use CollectionLoader for applying a batch fetch
+ *
  * @author Steve Ebersole
  */
 public class BatchKeyCollectionLoader implements CollectionLoader {

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

@@ -33,7 +33,6 @@
 import org.hibernate.sql.ast.spi.SqlAliasBaseManager;
 import org.hibernate.sql.ast.spi.SqlAstCreationContext;
 import org.hibernate.sql.ast.spi.SqlExpressionResolver;
-import org.hibernate.sql.ast.spi.SqlSelection;
 import org.hibernate.sql.ast.tree.expression.ColumnReference;
 import org.hibernate.sql.ast.tree.expression.Expression;
 import org.hibernate.sql.ast.tree.expression.JdbcParameter;
@@ -82,7 +81,7 @@ public class LoaderSelectBuilder {
 	 */
 	public static SelectStatement createSelect(
 			Loadable loadable,
-			List<ModelPart> partsToSelect,
+			List<? extends ModelPart> partsToSelect,
 			ModelPart restrictedPart,
 			DomainResult cachedDomainResult,
 			int numberOfKeysToLoad,
@@ -143,7 +142,7 @@ public static SelectStatement createSubSelectFetchSelect(
 
 	private final SqlAstCreationContext creationContext;
 	private final Loadable loadable;
-	private final List<ModelPart> partsToSelect;
+	private final List<? extends ModelPart> partsToSelect;
 	private final ModelPart restrictedPart;
 	private final DomainResult cachedDomainResult;
 	private final int numberOfKeysToLoad;
@@ -155,7 +154,7 @@ public static SelectStatement createSubSelectFetchSelect(
 	private LoaderSelectBuilder(
 			SqlAstCreationContext creationContext,
 			Loadable loadable,
-			List<ModelPart> partsToSelect,
+			List<? extends ModelPart> partsToSelect,
 			ModelPart restrictedPart,
 			DomainResult cachedDomainResult,
 			int numberOfKeysToLoad,

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

@@ -6,19 +6,53 @@
  */
 package org.hibernate.loader.ast.internal;
 
-import org.hibernate.NotYetImplementedFor6Exception;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+import org.hibernate.HibernateException;
+import org.hibernate.LockOptions;
+import org.hibernate.engine.jdbc.env.spi.JdbcEnvironment;
+import org.hibernate.engine.jdbc.spi.JdbcServices;
+import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
+import org.hibernate.internal.util.StringHelper;
 import org.hibernate.loader.ast.spi.NaturalIdLoader;
+import org.hibernate.metamodel.mapping.JdbcMapping;
+import org.hibernate.metamodel.mapping.NaturalIdMapping;
+import org.hibernate.metamodel.mapping.SingularAttributeMapping;
 import org.hibernate.persister.entity.EntityPersister;
+import org.hibernate.query.spi.QueryOptions;
+import org.hibernate.query.spi.QueryParameterBindings;
+import org.hibernate.sql.ast.Clause;
+import org.hibernate.sql.ast.SqlAstTranslatorFactory;
+import org.hibernate.sql.ast.tree.expression.JdbcParameter;
+import org.hibernate.sql.ast.tree.select.SelectStatement;
+import org.hibernate.sql.exec.internal.JdbcParameterBindingsImpl;
+import org.hibernate.sql.exec.spi.Callback;
+import org.hibernate.sql.exec.spi.ExecutionContext;
+import org.hibernate.sql.exec.spi.JdbcParameterBinding;
+import org.hibernate.sql.exec.spi.JdbcParameterBindings;
+import org.hibernate.sql.exec.spi.JdbcSelect;
 
 /**
  * @author Steve Ebersole
  */
 public class NaturalIdLoaderStandardImpl<T> implements NaturalIdLoader<T> {
 	private final EntityPersister entityDescriptor;
+	private	final NaturalIdMapping naturalIdMapping;
 
 	public NaturalIdLoaderStandardImpl(EntityPersister entityDescriptor) {
 		this.entityDescriptor = entityDescriptor;
+		this.naturalIdMapping = entityDescriptor.getNaturalIdMapping();
+
+		if ( ! entityDescriptor.hasNaturalIdentifier() ) {
+			throw new HibernateException( "Entity does not define natural-id : " + entityDescriptor.getEntityName() );
+		}
+
+		// todo (6.0) : account for nullable attributes that are part of the natural-id (is-null-or-equals)
+		// todo (6.0) : cache the SQL AST and JdbcParameter list
 	}
 
 	@Override
@@ -28,6 +62,283 @@ public EntityPersister getLoadable() {
 
 	@Override
 	public T load(Object naturalIdToLoad, LoadOptions options, SharedSessionContractImplementor session) {
-		throw new NotYetImplementedFor6Exception( getClass() );
+		final SessionFactoryImplementor sessionFactory = session.getFactory();
+
+		final List<JdbcParameter> jdbcParameters = new ArrayList<>();
+		final SelectStatement sqlSelect = LoaderSelectBuilder.createSelect(
+				entityDescriptor,
+				Collections.singletonList( entityDescriptor.getIdentifierMapping() ),
+				naturalIdMapping,
+				null,
+				1,
+				session.getLoadQueryInfluencers(),
+				LockOptions.READ,
+				jdbcParameters::add,
+				sessionFactory
+		);
+
+		final JdbcServices jdbcServices = sessionFactory.getJdbcServices();
+		final JdbcEnvironment jdbcEnvironment = jdbcServices.getJdbcEnvironment();
+		final SqlAstTranslatorFactory sqlAstTranslatorFactory = jdbcEnvironment.getSqlAstTranslatorFactory();
+
+		final JdbcSelect jdbcSelect = sqlAstTranslatorFactory.buildSelectTranslator( sessionFactory ).translate( sqlSelect );
+
+		final JdbcParameterBindings jdbcParamBindings = new JdbcParameterBindingsImpl( jdbcParameters.size() );
+		final Iterator<JdbcParameter> jdbcParamItr = jdbcParameters.iterator();
+
+		for ( int i = 0; i < naturalIdMapping.getNaturalIdAttributes().size(); i++ ) {
+			final SingularAttributeMapping attrMapping = naturalIdMapping.getNaturalIdAttributes().get( i );
+			attrMapping.visitJdbcValues(
+					naturalIdToLoad,
+					Clause.WHERE,
+					(jdbcValue, jdbcMapping) -> {
+						assert jdbcParamItr.hasNext();
+						final JdbcParameter jdbcParam = jdbcParamItr.next();
+						jdbcParamBindings.addBinding(
+								jdbcParam,
+								new JdbcParameterBinding() {
+									@Override
+									public JdbcMapping getBindType() {
+										return jdbcMapping;
+									}
+
+									@Override
+									public Object getBindValue() {
+										return jdbcValue;
+									}
+								}
+						);
+					},
+					session
+			);
+		}
+
+		final List<Object[]> results = session.getFactory().getJdbcServices().getJdbcSelectExecutor().list(
+				jdbcSelect,
+				jdbcParamBindings,
+				new ExecutionContext() {
+					@Override
+					public SharedSessionContractImplementor getSession() {
+						return session;
+					}
+
+					@Override
+					public QueryOptions getQueryOptions() {
+						return QueryOptions.NONE;
+					}
+
+					@Override
+					public QueryParameterBindings getQueryParameterBindings() {
+						return QueryParameterBindings.NO_PARAM_BINDINGS;
+					}
+
+					@Override
+					public Callback getCallback() {
+						return afterLoadAction -> {
+						};
+					}
+				},
+				row -> row
+		);
+
+		if ( results.size() > 1 ) {
+			throw new HibernateException(
+					String.format(
+							"Loading by natural-id returned more that one row : %s",
+							entityDescriptor.getEntityName()
+					)
+			);
+		}
+
+		//noinspection unchecked
+		return (T) results.get( 0 );
+	}
+
+	@Override
+	public Object[] resolveIdToNaturalId(Object id, SharedSessionContractImplementor session) {
+		final SessionFactoryImplementor sessionFactory = session.getFactory();
+
+		final List<JdbcParameter> jdbcParameters = new ArrayList<>();
+		final SelectStatement sqlSelect = LoaderSelectBuilder.createSelect(
+				entityDescriptor,
+				naturalIdMapping.getNaturalIdAttributes(),
+				entityDescriptor.getIdentifierMapping(),
+				null,
+				1,
+				session.getLoadQueryInfluencers(),
+				LockOptions.READ,
+				jdbcParameters::add,
+				sessionFactory
+		);
+
+		final JdbcServices jdbcServices = sessionFactory.getJdbcServices();
+		final JdbcEnvironment jdbcEnvironment = jdbcServices.getJdbcEnvironment();
+		final SqlAstTranslatorFactory sqlAstTranslatorFactory = jdbcEnvironment.getSqlAstTranslatorFactory();
+
+		final JdbcSelect jdbcSelect = sqlAstTranslatorFactory.buildSelectTranslator( sessionFactory ).translate( sqlSelect );
+
+		final JdbcParameterBindings jdbcParamBindings = new JdbcParameterBindingsImpl( jdbcParameters.size() );
+		final Iterator<JdbcParameter> jdbcParamItr = jdbcParameters.iterator();
+
+		entityDescriptor.getIdentifierMapping().visitJdbcValues(
+				id,
+				Clause.WHERE,
+				(value, type) -> {
+					assert jdbcParamItr.hasNext();
+					final JdbcParameter jdbcParam = jdbcParamItr.next();
+					jdbcParamBindings.addBinding(
+							jdbcParam,
+							new JdbcParameterBinding() {
+								@Override
+								public JdbcMapping getBindType() {
+									return type;
+								}
+
+								@Override
+								public Object getBindValue() {
+									return value;
+								}
+							}
+					);
+				},
+				session
+		);
+
+
+		final List<Object[]> results = session.getFactory().getJdbcServices().getJdbcSelectExecutor().list(
+				jdbcSelect,
+				jdbcParamBindings,
+				new ExecutionContext() {
+					@Override
+					public SharedSessionContractImplementor getSession() {
+						return session;
+					}
+
+					@Override
+					public QueryOptions getQueryOptions() {
+						return QueryOptions.NONE;
+					}
+
+					@Override
+					public QueryParameterBindings getQueryParameterBindings() {
+						return QueryParameterBindings.NO_PARAM_BINDINGS;
+					}
+
+					@Override
+					public Callback getCallback() {
+						return afterLoadAction -> {
+						};
+					}
+				},
+				row -> row
+		);
+
+		if ( results.size() > 1 ) {
+			throw new HibernateException(
+					String.format(
+							"Resolving id to natural-id returned more that one row : %s #%s",
+							entityDescriptor.getEntityName(),
+							id
+					)
+			);
+		}
+
+		return results.get( 0 );
+	}
+
+	@Override
+	public Object resolveNaturalIdToId(
+			Object[] naturalIdValues,
+			SharedSessionContractImplementor session) {
+		final SessionFactoryImplementor sessionFactory = session.getFactory();
+
+		final List<JdbcParameter> jdbcParameters = new ArrayList<>();
+		final SelectStatement sqlSelect = LoaderSelectBuilder.createSelect(
+				entityDescriptor,
+				Collections.emptyList(),
+				naturalIdMapping,
+				null,
+				1,
+				session.getLoadQueryInfluencers(),
+				LockOptions.READ,
+				jdbcParameters::add,
+				sessionFactory
+		);
+
+		final JdbcServices jdbcServices = sessionFactory.getJdbcServices();
+		final JdbcEnvironment jdbcEnvironment = jdbcServices.getJdbcEnvironment();
+		final SqlAstTranslatorFactory sqlAstTranslatorFactory = jdbcEnvironment.getSqlAstTranslatorFactory();
+
+		final JdbcSelect jdbcSelect = sqlAstTranslatorFactory.buildSelectTranslator( sessionFactory ).translate( sqlSelect );
+
+		final JdbcParameterBindings jdbcParamBindings = new JdbcParameterBindingsImpl( jdbcParameters.size() );
+		final Iterator<JdbcParameter> jdbcParamItr = jdbcParameters.iterator();
+
+		for ( int i = 0; i < naturalIdMapping.getNaturalIdAttributes().size(); i++ ) {
+			final SingularAttributeMapping attrMapping = naturalIdMapping.getNaturalIdAttributes().get( i );
+			attrMapping.visitJdbcValues(
+					naturalIdValues[i],
+					Clause.WHERE,
+					(jdbcValue, jdbcMapping) -> {
+						assert jdbcParamItr.hasNext();
+						jdbcParamBindings.addBinding(
+								jdbcParamItr.next(),
+								new JdbcParameterBinding() {
+									@Override
+									public JdbcMapping getBindType() {
+										return jdbcMapping;
+									}
+
+									@Override
+									public Object getBindValue() {
+										return jdbcValue;
+									}
+								}
+						);
+					},
+					session
+			);
+		}
+		assert !jdbcParamItr.hasNext();
+
+		final List<Object[]> results = session.getFactory().getJdbcServices().getJdbcSelectExecutor().list(
+				jdbcSelect,
+				jdbcParamBindings,
+				new ExecutionContext() {
+					@Override
+					public SharedSessionContractImplementor getSession() {
+						return session;
+					}
+
+					@Override
+					public QueryOptions getQueryOptions() {
+						return QueryOptions.NONE;
+					}
+
+					@Override
+					public QueryParameterBindings getQueryParameterBindings() {
+						return QueryParameterBindings.NO_PARAM_BINDINGS;
+					}
+
+					@Override
+					public Callback getCallback() {
+						return afterLoadAction -> {
+						};
+					}
+				},
+				row -> row
+		);
+
+		if ( results.size() > 1 ) {
+			throw new HibernateException(
+					String.format(
+							"Resolving natural-id to id returned more that one row : %s [%s]",
+							entityDescriptor.getEntityName(),
+							StringHelper.join( ", ", naturalIdValues )
+					)
+			);
+		}
+
+		return results.get( 0 );
 	}
 }

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

@@ -6,8 +6,6 @@
  */
 package org.hibernate.loader.ast.internal;
 
-import java.util.ArrayList;
-
 import org.hibernate.LockOptions;
 import org.hibernate.collection.spi.PersistentCollection;
 import org.hibernate.engine.jdbc.env.spi.JdbcEnvironment;
@@ -24,10 +22,9 @@
 import org.hibernate.sql.ast.tree.select.SelectStatement;
 import org.hibernate.sql.exec.spi.Callback;
 import org.hibernate.sql.exec.spi.ExecutionContext;
-import org.hibernate.sql.ast.tree.expression.JdbcParameter;
 import org.hibernate.sql.exec.spi.JdbcSelect;
-import org.hibernate.sql.results.internal.RowTransformerPassThruImpl;
 import org.hibernate.sql.results.graph.DomainResult;
+import org.hibernate.sql.results.internal.RowTransformerPassThruImpl;
 
 /**
  * A one-time use CollectionLoader for applying a sub-select fetch
@@ -36,11 +33,9 @@
  */
 public class SubSelectFetchCollectionLoader implements CollectionLoader {
 	private final PluralAttributeMapping attributeMapping;
-	private final DomainResult cachedDomainResult;
 	private final SubselectFetch subselect;
 
 	private final SelectStatement sqlAst;
-	private final java.util.List<JdbcParameter> jdbcParameters;
 
 	public SubSelectFetchCollectionLoader(
 			PluralAttributeMapping attributeMapping,
@@ -48,18 +43,15 @@ public SubSelectFetchCollectionLoader(
 			SubselectFetch subselect,
 			SharedSessionContractImplementor session) {
 		this.attributeMapping = attributeMapping;
-		this.cachedDomainResult = cachedDomainResult;
 		this.subselect = subselect;
 
-		jdbcParameters = new ArrayList<>();
-
 		sqlAst = LoaderSelectBuilder.createSubSelectFetchSelect(
 				attributeMapping,
 				subselect,
-				null,
+				cachedDomainResult,
 				session.getLoadQueryInfluencers(),
 				LockOptions.READ,
-				jdbcParameters::add,
+				jdbcParameter -> {},
 				session.getFactory()
 		);
 	}

hibernate-core/src/main/java/org/hibernate/loader/ast/spi/NaturalIdLoader.java

@@ -42,4 +42,11 @@ interface LoadOptions {
 	 * @param session The session into which the entity is being loaded
 	 */
 	T load(Object naturalIdToLoad, LoadOptions options, SharedSessionContractImplementor session);
+
+	/**
+	 * Resolve the natural-id value from an id
+	 */
+	Object[] resolveIdToNaturalId(Object id, SharedSessionContractImplementor session);
+
+	Object resolveNaturalIdToId(Object[] naturalIdValues, SharedSessionContractImplementor session);
 }

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

@@ -158,8 +158,8 @@ default void visitJdbcValues(
 	@FunctionalInterface
 	interface JdbcValuesConsumer {
 		/**
-		 * Consume a JDBC-level value.  The JDBC type descriptor is also passed in
+		 * Consume a JDBC-level jdbcValue.  The JDBC jdbcMapping descriptor is also passed in
 		 */
-		void consume(Object value, JdbcMapping type);
+		void consume(Object jdbcValue, JdbcMapping jdbcMapping);
 	}
 }

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

@@ -6,8 +6,11 @@
  */
 package org.hibernate.metamodel.mapping;
 
+import java.util.List;
+
 /**
  * @author Steve Ebersole
  */
-public interface NaturalIdMapping extends SingularAttributeMapping, StateArrayContributorMapping {
+public interface NaturalIdMapping extends VirtualModelPart {
+	List<SingularAttributeMapping> getNaturalIdAttributes();
 }

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

@@ -21,7 +21,6 @@
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import java.util.SortedMap;
@@ -74,7 +73,6 @@
 import org.hibernate.engine.internal.Versioning;
 import org.hibernate.engine.jdbc.batch.internal.BasicBatchKey;
 import org.hibernate.engine.jdbc.env.spi.JdbcEnvironment;
-import org.hibernate.engine.jdbc.spi.JdbcCoordinator;
 import org.hibernate.engine.jdbc.spi.JdbcServices;
 import org.hibernate.engine.spi.CachedNaturalIdValueSource;
 import org.hibernate.engine.spi.CascadeStyle;
@@ -107,11 +105,6 @@
 import org.hibernate.jdbc.Expectation;
 import org.hibernate.jdbc.Expectations;
 import org.hibernate.jdbc.TooManyRowsAffectedException;
-import org.hibernate.loader.custom.sql.SQLQueryParser;
-import org.hibernate.loader.entity.BatchingEntityLoaderBuilder;
-import org.hibernate.loader.entity.CascadeEntityLoader;
-import org.hibernate.loader.entity.EntityLoader;
-import org.hibernate.loader.entity.UniqueEntityLoader;
 import org.hibernate.loader.ast.internal.MultiIdEntityLoaderStandardImpl;
 import org.hibernate.loader.ast.internal.NaturalIdLoaderStandardImpl;
 import org.hibernate.loader.ast.internal.Preparable;
@@ -122,6 +115,11 @@
 import org.hibernate.loader.ast.spi.MultiIdEntityLoader;
 import org.hibernate.loader.ast.spi.NaturalIdLoader;
 import org.hibernate.loader.ast.spi.SingleIdEntityLoader;
+import org.hibernate.loader.custom.sql.SQLQueryParser;
+import org.hibernate.loader.entity.BatchingEntityLoaderBuilder;
+import org.hibernate.loader.entity.CascadeEntityLoader;
+import org.hibernate.loader.entity.EntityLoader;
+import org.hibernate.loader.entity.UniqueEntityLoader;
 import org.hibernate.mapping.Column;
 import org.hibernate.mapping.Component;
 import org.hibernate.mapping.Formula;
@@ -194,11 +192,11 @@
 import org.hibernate.sql.ast.tree.predicate.ComparisonPredicate;
 import org.hibernate.sql.ast.tree.predicate.Junction;
 import org.hibernate.sql.ast.tree.predicate.Predicate;
-import org.hibernate.sql.results.graph.entity.internal.EntityResultImpl;
 import org.hibernate.sql.results.graph.DomainResult;
 import org.hibernate.sql.results.graph.DomainResultCreationState;
 import org.hibernate.sql.results.graph.Fetchable;
 import org.hibernate.sql.results.graph.FetchableContainer;
+import org.hibernate.sql.results.graph.entity.internal.EntityResultImpl;
 import org.hibernate.stat.spi.StatisticsImplementor;
 import org.hibernate.tuple.GenerationTiming;
 import org.hibernate.tuple.InDatabaseValueGenerationStrategy;
@@ -4527,7 +4525,7 @@ public final void postInstantiate() throws MappingException {
 		prepareLoader( multiIdEntityLoader );
 		prepareLoader( naturalIdLoader );
 
-		// todo (6.0) : the init done in most of these is delayed now
+		// todo (6.0) : these should be removed in favor of `singleIdEntityLoader`, ...
 		createLoaders();
 		createUniqueKeyLoaders();
 
@@ -4580,49 +4578,6 @@ protected void createLoaders() {
 		);
 	}
 
-	protected final UniqueEntityLoader getLoaderByLockMode(LockMode lockMode) {
-		if ( LockMode.NONE == lockMode ) {
-			return noneLockLoader;
-		}
-		else if ( LockMode.READ == lockMode ) {
-			return readLockLoader;
-		}
-
-		return loaders.computeIfAbsent( lockMode, this::generateDelayedEntityLoader );
-	}
-
-	private UniqueEntityLoader generateDelayedEntityLoader(Object lockModeObject) {
-		// Unfortunately, the loaders map mixes LockModes and Strings as keys so we need to accept an Object.
-		// The cast is safe as we will always call this method with a LockMode.
-		LockMode lockMode = (LockMode) lockModeObject;
-
-		switch ( lockMode ) {
-			case NONE:
-			case READ:
-			case OPTIMISTIC:
-			case OPTIMISTIC_FORCE_INCREMENT: {
-				return createEntityLoader( lockMode );
-			}
-			case UPGRADE:
-			case UPGRADE_NOWAIT:
-			case UPGRADE_SKIPLOCKED:
-			case FORCE:
-			case PESSIMISTIC_READ:
-			case PESSIMISTIC_WRITE:
-			case PESSIMISTIC_FORCE_INCREMENT: {
-				//TODO: inexact, what we really need to know is: are any outer joins used?
-				boolean disableForUpdate = getSubclassTableSpan() > 1
-						&& hasSubclasses()
-						&& !getFactory().getDialect().supportsOuterJoinForUpdate();
-
-				return disableForUpdate ? readLockLoader : createEntityLoader( lockMode );
-			}
-			default: {
-				throw new IllegalStateException( String.format( Locale.ROOT, "Lock mode %1$s not supported by entity loaders.", lockMode ) );
-			}
-		}
-	}
-
 	/**
 	 * Load an instance using either the <tt>forUpdateLoader</tt> or the outer joining <tt>loader</tt>,
 	 * depending upon the value of the <tt>lock</tt> parameter
@@ -5667,109 +5622,33 @@ public int[] getNaturalIdentifierProperties() {
 		return entityMetamodel.getNaturalIdentifierProperties();
 	}
 
-	public Object[] getNaturalIdentifierSnapshot(Object id, SharedSessionContractImplementor session)
-			throws HibernateException {
-		if ( !hasNaturalIdentifier() ) {
-			throw new MappingException(
-					"persistent class did not define a natural-id : " + MessageHelper.infoString(
-							this
-					)
-			);
-		}
+	public Object[] getNaturalIdentifierSnapshot(Object id, SharedSessionContractImplementor session) {
+		verifyHasNaturalId();
+
 		if ( LOG.isTraceEnabled() ) {
-			LOG.tracev(
-					"Getting current natural-id snapshot state for: {0}",
-					MessageHelper.infoString( this, id, getFactory() )
+			LOG.tracef(
+					"Getting current natural-id snapshot state for `%s#%s",
+					getEntityName(),
+					id
 			);
 		}
 
-		int[] naturalIdPropertyIndexes = getNaturalIdentifierProperties();
-		int naturalIdPropertyCount = naturalIdPropertyIndexes.length;
-		boolean[] naturalIdMarkers = new boolean[getPropertySpan()];
-		Type[] extractionTypes = new Type[naturalIdPropertyCount];
-		for ( int i = 0; i < naturalIdPropertyCount; i++ ) {
-			extractionTypes[i] = getPropertyTypes()[naturalIdPropertyIndexes[i]];
-			naturalIdMarkers[naturalIdPropertyIndexes[i]] = true;
-		}
+		return naturalIdLoader.resolveIdToNaturalId( id, session );
+	}
 
-		///////////////////////////////////////////////////////////////////////
-		// TODO : look at perhaps caching this...
-		Select select = new Select( getFactory().getDialect() );
-		if ( getFactory().getSessionFactoryOptions().isCommentsEnabled() ) {
-			select.setComment( "get current natural-id state " + getEntityName() );
-		}
-		select.setSelectClause( concretePropertySelectFragmentSansLeadingComma( getRootAlias(), naturalIdMarkers ) );
-		select.setFromClause( fromTableFragment( getRootAlias() ) + fromJoinFragment( getRootAlias(), true, false ) );
-
-		String[] aliasedIdColumns = StringHelper.qualify( getRootAlias(), getIdentifierColumnNames() );
-		String whereClause = new StringBuilder()
-				.append(
-						String.join(
-								"=? and ",
-								aliasedIdColumns
-						)
-				)
-				.append( "=?" )
-				.append( whereJoinFragment( getRootAlias(), true, false ) )
-				.toString();
-
-		String sql = select.setOuterJoins( "", "" )
-				.setWhereClause( whereClause )
-				.toStatementString();
-		///////////////////////////////////////////////////////////////////////
-
-		Object[] snapshot = new Object[naturalIdPropertyCount];
-		try {
-			final JdbcCoordinator jdbcCoordinator = session.getJdbcCoordinator();
-			PreparedStatement ps = jdbcCoordinator
-					.getStatementPreparer()
-					.prepareStatement( sql );
-			try {
-				getIdentifierType().nullSafeSet( ps, id, 1, session );
-				ResultSet rs = jdbcCoordinator.getResultSetReturn().extract( ps );
-				try {
-					//if there is no resulting row, return null
-					if ( !rs.next() ) {
-						return null;
-					}
-					final EntityKey key = session.generateEntityKey( id, this );
-					Object owner = session.getPersistenceContextInternal().getEntity( key );
-					for ( int i = 0; i < naturalIdPropertyCount; i++ ) {
-						snapshot[i] = extractionTypes[i].hydrate(
-								rs, getPropertyAliases(
-										"",
-										naturalIdPropertyIndexes[i]
-								), session, null
-						);
-						if ( extractionTypes[i].isEntityType() ) {
-							snapshot[i] = extractionTypes[i].resolve( snapshot[i], session, owner );
-						}
-					}
-					return snapshot;
-				}
-				finally {
-					jdbcCoordinator.getResourceRegistry().release( rs, ps );
-				}
-			}
-			finally {
-				jdbcCoordinator.getResourceRegistry().release( ps );
-				jdbcCoordinator.afterStatementExecution();
-			}
-		}
-		catch (SQLException e) {
-			throw getFactory().getSQLExceptionHelper().convert(
-					e,
-					"could not retrieve snapshot: " + MessageHelper.infoString( this, id, getFactory() ),
-					sql
-			);
+	private void verifyHasNaturalId() {
+		if ( ! hasNaturalIdentifier() ) {
+			throw new HibernateException( "Entity does not define a natural id : " + getEntityName() );
 		}
 	}
 
 	@Override
-	public Serializable loadEntityIdByNaturalId(
+	public Object loadEntityIdByNaturalId(
 			Object[] naturalIdValues,
 			LockOptions lockOptions,
 			SharedSessionContractImplementor session) {
+		verifyHasNaturalId();
+
 		if ( LOG.isTraceEnabled() ) {
 			LOG.tracef(
 					"Resolving natural-id [%s] to id : %s ",
@@ -5778,162 +5657,7 @@ public Serializable loadEntityIdByNaturalId(
 			);
 		}
 
-		final boolean[] valueNullness = determineValueNullness( naturalIdValues );
-		final String sqlEntityIdByNaturalIdString = determinePkByNaturalIdQuery( valueNullness );
-
-		try {
-			PreparedStatement ps = session
-					.getJdbcCoordinator()
-					.getStatementPreparer()
-					.prepareStatement( sqlEntityIdByNaturalIdString );
-			try {
-				int positions = 1;
-				int loop = 0;
-				for ( int idPosition : getNaturalIdentifierProperties() ) {
-					final Object naturalIdValue = naturalIdValues[loop++];
-					if ( naturalIdValue != null ) {
-						final Type type = getPropertyTypes()[idPosition];
-						type.nullSafeSet( ps, naturalIdValue, positions, session );
-						positions += type.getColumnSpan( session.getFactory() );
-					}
-				}
-				ResultSet rs = session.getJdbcCoordinator().getResultSetReturn().extract( ps );
-				try {
-					// if there is no resulting row, return null
-					if ( !rs.next() ) {
-						return null;
-					}
-
-					final Object hydratedId = getIdentifierType().hydrate( rs, getIdentifierAliases(), session, null );
-					return (Serializable) getIdentifierType().resolve( hydratedId, session, null );
-				}
-				finally {
-					session.getJdbcCoordinator().getResourceRegistry().release( rs, ps );
-				}
-			}
-			finally {
-				session.getJdbcCoordinator().getResourceRegistry().release( ps );
-				session.getJdbcCoordinator().afterStatementExecution();
-			}
-		}
-		catch (SQLException e) {
-			throw getFactory().getSQLExceptionHelper().convert(
-					e,
-					String.format(
-							"could not resolve natural-id [%s] to id : %s",
-							Arrays.asList( naturalIdValues ),
-							MessageHelper.infoString( this )
-					),
-					sqlEntityIdByNaturalIdString
-			);
-		}
-	}
-
-	private boolean[] determineValueNullness(Object[] naturalIdValues) {
-		boolean[] nullness = new boolean[naturalIdValues.length];
-		for ( int i = 0; i < naturalIdValues.length; i++ ) {
-			nullness[i] = naturalIdValues[i] == null;
-		}
-		return nullness;
-	}
-
-	private Boolean naturalIdIsNonNullable;
-	private String cachedPkByNonNullableNaturalIdQuery;
-
-	private String determinePkByNaturalIdQuery(boolean[] valueNullness) {
-		if ( !hasNaturalIdentifier() ) {
-			throw new HibernateException(
-					"Attempt to build natural-id -> PK resolution query for entity that does not define natural id"
-			);
-		}
-
-		// performance shortcut for cases where the natural-id is defined as completely non-nullable
-		if ( isNaturalIdNonNullable() ) {
-			if ( valueNullness != null && !ArrayHelper.isAllFalse( valueNullness ) ) {
-				throw new HibernateException( "Null value(s) passed to lookup by non-nullable natural-id" );
-			}
-			if ( cachedPkByNonNullableNaturalIdQuery == null ) {
-				cachedPkByNonNullableNaturalIdQuery = generateEntityIdByNaturalIdSql( null );
-			}
-			return cachedPkByNonNullableNaturalIdQuery;
-		}
-
-		// Otherwise, regenerate it each time
-		return generateEntityIdByNaturalIdSql( valueNullness );
-	}
-
-	protected boolean isNaturalIdNonNullable() {
-		if ( naturalIdIsNonNullable == null ) {
-			naturalIdIsNonNullable = determineNaturalIdNullability();
-		}
-		return naturalIdIsNonNullable;
-	}
-
-	private boolean determineNaturalIdNullability() {
-		boolean[] nullability = getPropertyNullability();
-		for ( int position : getNaturalIdentifierProperties() ) {
-			// if any individual property is nullable, return false
-			if ( nullability[position] ) {
-				return false;
-			}
-		}
-		// return true if we found no individually nullable properties
-		return true;
-	}
-
-	private String generateEntityIdByNaturalIdSql(boolean[] valueNullness) {
-		EntityPersister rootPersister = getFactory().getEntityPersister( getRootEntityName() );
-		if ( rootPersister != this ) {
-			if ( rootPersister instanceof AbstractEntityPersister ) {
-				return ( (AbstractEntityPersister) rootPersister ).generateEntityIdByNaturalIdSql( valueNullness );
-			}
-		}
-
-		Select select = new Select( getFactory().getDialect() );
-		if ( getFactory().getSessionFactoryOptions().isCommentsEnabled() ) {
-			select.setComment( "get current natural-id->entity-id state " + getEntityName() );
-		}
-
-		final String rootAlias = getRootAlias();
-
-		select.setSelectClause( identifierSelectFragment( rootAlias, "" ) );
-		select.setFromClause( fromTableFragment( rootAlias ) + fromJoinFragment( rootAlias, true, false ) );
-
-		final StringBuilder whereClause = new StringBuilder();
-		final int[] propertyTableNumbers = getPropertyTableNumbers();
-		final int[] naturalIdPropertyIndexes = this.getNaturalIdentifierProperties();
-		int valuesIndex = -1;
-		for ( int propIdx = 0; propIdx < naturalIdPropertyIndexes.length; propIdx++ ) {
-			valuesIndex++;
-			if ( propIdx > 0 ) {
-				whereClause.append( " and " );
-			}
-
-			final int naturalIdIdx = naturalIdPropertyIndexes[propIdx];
-			final String tableAlias = generateTableAlias( rootAlias, propertyTableNumbers[naturalIdIdx] );
-			final String[] propertyColumnNames = getPropertyColumnNames( naturalIdIdx );
-			final String[] aliasedPropertyColumns = StringHelper.qualify( tableAlias, propertyColumnNames );
-
-			if ( valueNullness != null && valueNullness[valuesIndex] ) {
-				whereClause.append( String.join( " is null and ", aliasedPropertyColumns ) ).append( " is null" );
-			}
-			else {
-				whereClause.append( String.join( "=? and ", aliasedPropertyColumns ) ).append( "=?" );
-			}
-		}
-
-		whereClause.append( whereJoinFragment( getRootAlias(), true, false ) );
-
-		return select.setOuterJoins( "", "" ).setWhereClause( whereClause.toString() ).toStatementString();
-	}
-
-	protected String concretePropertySelectFragmentSansLeadingComma(String alias, boolean[] include) {
-		String concretePropertySelectFragment = concretePropertySelectFragment( alias, include );
-		int firstComma = concretePropertySelectFragment.indexOf( ", " );
-		if ( firstComma == 0 ) {
-			concretePropertySelectFragment = concretePropertySelectFragment.substring( 2 );
-		}
-		return concretePropertySelectFragment;
+		return naturalIdLoader.resolveNaturalIdToId( naturalIdValues, session );
 	}
 
 	public boolean hasNaturalIdentifier() {

hibernate-core/src/main/java/org/hibernate/query/sqm/internal/SqmUtil.java

@@ -255,7 +255,7 @@ private static void createValueBindings(
 					private int position = 0;
 
 					@Override
-					public void consume(Object jdbcValue, JdbcMapping jdbcType) {
+					public void consume(Object jdbcValue, JdbcMapping jdbcMapping) {
 						final JdbcParameter jdbcParameter = jdbcParams.get( position );
 						jdbcParameterBindings.addBinding(
 								jdbcParameter,
@@ -263,7 +263,7 @@ public void consume(Object jdbcValue, JdbcMapping jdbcType) {
 
 									@Override
 									public JdbcMapping getBindType() {
-										return jdbcType;
+										return jdbcMapping;
 									}
 
 									@Override