diff --git a/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/AbstractCollectionBatchLoader.java b/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/AbstractCollectionBatchLoader.java
index 7ccec36405..c37b5a480d 100644
--- a/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/AbstractCollectionBatchLoader.java
+++ b/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/AbstractCollectionBatchLoader.java
@@ -14,7 +14,10 @@ import org.hibernate.engine.spi.PersistenceContext;
 import org.hibernate.engine.spi.SessionFactoryImplementor;
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.loader.ast.spi.CollectionBatchLoader;
+import org.hibernate.metamodel.mapping.NonAggregatedIdentifierMapping;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
+import org.hibernate.metamodel.mapping.ValuedModelPart;
+import org.hibernate.metamodel.mapping.internal.IdClassEmbeddable;
 import org.hibernate.sql.results.internal.ResultsHelper;
 
 import java.lang.reflect.Array;
@@ -127,8 +130,10 @@ public abstract class AbstractCollectionBatchLoader implements CollectionBatchLo
 
 	Object[] resolveKeysToInitialize(Object keyBeingLoaded, SharedSessionContractImplementor session) {
 		final int length = getDomainBatchSize();
-		final Class<?> keyType = getLoadable().getKeyDescriptor().getJavaType().getJavaTypeClass();
-		final Object[] keysToInitialize = (Object[]) Array.newInstance( keyType, length );
+		final Object[] keysToInitialize = (Object[]) Array.newInstance(
+				getKeyType( getLoadable().getKeyDescriptor().getKeyPart() ),
+				length
+		);
 		session.getPersistenceContextInternal().getBatchFetchQueue()
 				.collectBatchLoadableCollectionKeys(
 						length,
@@ -140,4 +145,15 @@ public abstract class AbstractCollectionBatchLoader implements CollectionBatchLo
 		return trimIdBatch( length, keysToInitialize );
 	}
 
+	protected Class<?> getKeyType(ValuedModelPart keyPart) {
+		if ( keyPart instanceof NonAggregatedIdentifierMapping ) {
+			final IdClassEmbeddable idClassEmbeddable = ( (NonAggregatedIdentifierMapping) keyPart ).getIdClassEmbeddable();
+			if ( idClassEmbeddable != null ) {
+				return idClassEmbeddable.getMappedJavaType().getJavaTypeClass();
+			}
+		}
+		return keyPart.getJavaType().getJavaTypeClass();
+	}
+
+
 }
diff --git a/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/CollectionBatchLoaderArrayParam.java b/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/CollectionBatchLoaderArrayParam.java
index 5f3bd9923d..3c0e5cee3e 100644
--- a/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/CollectionBatchLoaderArrayParam.java
+++ b/hibernate-core/src/main/java/org/hibernate/loader/ast/internal/CollectionBatchLoaderArrayParam.java
@@ -19,7 +19,10 @@ import org.hibernate.loader.ast.spi.CollectionBatchLoader;
 import org.hibernate.loader.ast.spi.SqlArrayMultiKeyLoader;
 import org.hibernate.metamodel.mapping.ForeignKeyDescriptor;
 import org.hibernate.metamodel.mapping.JdbcMapping;
+import org.hibernate.metamodel.mapping.NonAggregatedIdentifierMapping;
 import org.hibernate.metamodel.mapping.PluralAttributeMapping;
+import org.hibernate.metamodel.mapping.ValuedModelPart;
+import org.hibernate.metamodel.mapping.internal.IdClassEmbeddable;
 import org.hibernate.query.spi.QueryOptions;
 import org.hibernate.sql.ast.tree.expression.JdbcParameter;
 import org.hibernate.sql.ast.tree.select.SelectStatement;
@@ -45,6 +48,7 @@ import static org.hibernate.loader.ast.internal.MultiKeyLoadLogging.MULTI_KEY_LO
 public class CollectionBatchLoaderArrayParam
 		extends AbstractCollectionBatchLoader
 		implements SqlArrayMultiKeyLoader {
+	private final Class<?> keyDomainType;
 	private final JdbcMapping arrayJdbcMapping;
 	private final JdbcParameter jdbcParameter;
 	private final SelectStatement sqlSelect;
@@ -66,16 +70,18 @@ public class CollectionBatchLoaderArrayParam
 		}
 
 		final ForeignKeyDescriptor keyDescriptor = getLoadable().getKeyDescriptor();
-		final Class<?> keyType = keyDescriptor.getJavaType().getJavaTypeClass();
-		final Class<?> arrayClass = Array.newInstance( keyType, 0 ).getClass();
+		final JdbcMapping jdbcMapping = keyDescriptor.getSingleJdbcMapping();
+		final Class<?> jdbcArrayClass = Array.newInstance( jdbcMapping.getJdbcJavaType().getJavaTypeClass(), 0 )
+				.getClass();
+		keyDomainType = getKeyType( keyDescriptor.getKeyPart() );
 
 		final BasicType<?> arrayBasicType = getSessionFactory().getTypeConfiguration()
 				.getBasicTypeRegistry()
-				.getRegisteredType( arrayClass );
+				.getRegisteredType( jdbcArrayClass );
 		arrayJdbcMapping = MultiKeyLoadHelper.resolveArrayJdbcMapping(
 				arrayBasicType,
-				keyDescriptor.getSingleJdbcMapping(),
-				arrayClass,
+				jdbcMapping,
+				jdbcArrayClass,
 				getSessionFactory()
 		);
 
@@ -95,7 +101,6 @@ public class CollectionBatchLoaderArrayParam
 				.buildSelectTranslator( getSessionFactory(), sqlSelect )
 				.translate( JdbcParameterBindings.NO_BINDINGS, QueryOptions.NONE );
 	}
-
 	@Override
 	public PersistentCollection<?> load(Object keyBeingLoaded, SharedSessionContractImplementor session) {
 		final ForeignKeyDescriptor keyDescriptor = getLoadable().getKeyDescriptor();
@@ -122,13 +127,14 @@ public class CollectionBatchLoaderArrayParam
 
 		final int length = getDomainBatchSize();
 		final Object[] keysToInitialize = (Object[]) Array.newInstance(
-				keyDescriptor.getSingleJdbcMapping().getJdbcJavaType().getJavaTypeClass(),
-				length
-		);
-		final Object[] embeddedKeys = (Object[]) Array.newInstance(
-				keyDescriptor.getJavaType().getJavaTypeClass(),
+				jdbcParameter.getExpressionType()
+						.getSingleJdbcMapping()
+						.getJdbcJavaType()
+						.getJavaTypeClass()
+						.getComponentType(),
 				length
 		);
+		final Object[] embeddedKeys = (Object[]) Array.newInstance( keyDomainType, length );
 		session.getPersistenceContextInternal().getBatchFetchQueue()
 				.collectBatchLoadableCollectionKeys(
 						length,
@@ -157,7 +163,7 @@ public class CollectionBatchLoaderArrayParam
 		}
 		final CollectionKey collectionKey = new CollectionKey(
 				getLoadable().getCollectionDescriptor(),
-				keysToInitialize
+				keyBeingLoaded
 		);
 		return session.getPersistenceContext().getCollection( collectionKey );
 	}