Map collection and fields for $lookup/$graphLookup aggregation stage against domain type

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>4.2.0-SNAPSHOT</version>
+	<version>4.2.x-4379-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.x-4379-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.x-4379-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.x-4379-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/AggregationOperationContext.java

@@ -23,6 +23,7 @@
 import org.bson.codecs.configuration.CodecRegistry;
 import org.springframework.beans.BeanUtils;
 import org.springframework.data.mongodb.CodecRegistryProvider;
+import org.springframework.data.mongodb.MongoCollectionUtils;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -79,7 +80,30 @@ default Document getMappedObject(Document document) {
 	FieldReference getReference(String name);
 
 	/**
-	 * Returns the {@link Fields} exposed by the type. May be a {@literal class} or an {@literal interface}. The default
+	 * Obtain the target field name for a given field/type combination.
+	 *
+	 * @param type The type containing the field.
+	 * @param field The property/field name
+	 * @return never {@literal null}.
+	 * @since 4.2
+	 */
+	default String getMappedFieldName(Class<?> type, String field) {
+		return field;
+	}
+
+	/**
+	 * Obtain the collection name for a given {@link Class type} combination.
+	 *
+	 * @param type
+	 * @return never {@literal null}.
+	 * @since 4.2
+	 */
+	default String getCollection(Class<?> type) {
+		return MongoCollectionUtils.getPreferredCollectionName(type);
+	}
+
+	/**
+	 * Returns the {@link Fields} exposed by the type. Can be a {@literal class} or an {@literal interface}. The default
 	 * implementation uses {@link BeanUtils#getPropertyDescriptors(Class) property descriptors} discover fields from a
 	 * {@link Class}.
 	 *
@@ -109,7 +133,7 @@ default Fields getFields(Class<?> type) {
 
 	/**
 	 * This toggle allows the {@link AggregationOperationContext context} to use any given field name without checking for
-	 * its existence. Typically the {@link AggregationOperationContext} fails when referencing unknown fields, those that
+	 * its existence. Typically, the {@link AggregationOperationContext} fails when referencing unknown fields, those that
 	 * are not present in one of the previous stages or the input source, throughout the pipeline.
 	 *
 	 * @return a more relaxed {@link AggregationOperationContext}.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperation.java

@@ -46,7 +46,7 @@ public class GraphLookupOperation implements InheritsFieldsAggregationOperation
 	private static final Set<Class<?>> ALLOWED_START_TYPES = new HashSet<Class<?>>(
 			Arrays.<Class<?>> asList(AggregationExpression.class, String.class, Field.class, Document.class));
 
-	private final String from;
+	private final Object from;
 	private final List<Object> startWith;
 	private final Field connectFrom;
 	private final Field connectTo;
@@ -55,7 +55,7 @@ public class GraphLookupOperation implements InheritsFieldsAggregationOperation
 	private final @Nullable Field depthField;
 	private final @Nullable CriteriaDefinition restrictSearchWithMatch;
 
-	private GraphLookupOperation(String from, List<Object> startWith, Field connectFrom, Field connectTo, Field as,
+	private GraphLookupOperation(Object from, List<Object> startWith, Field connectFrom, Field connectTo, Field as,
 			@Nullable Long maxDepth, @Nullable Field depthField, @Nullable CriteriaDefinition restrictSearchWithMatch) {
 
 		this.from = from;
@@ -82,7 +82,7 @@ public Document toDocument(AggregationOperationContext context) {
 
 		Document graphLookup = new Document();
 
-		graphLookup.put("from", from);
+		graphLookup.put("from", getCollectionName(context));
 
 		List<Object> mappedStartWith = new ArrayList<>(startWith.size());
 
@@ -99,7 +99,7 @@ public Document toDocument(AggregationOperationContext context) {
 
 		graphLookup.put("startWith", mappedStartWith.size() == 1 ? mappedStartWith.iterator().next() : mappedStartWith);
 
-		graphLookup.put("connectFromField", connectFrom.getTarget());
+		graphLookup.put("connectFromField", getForeignFieldName(context));
 		graphLookup.put("connectToField", connectTo.getTarget());
 		graphLookup.put("as", as.getName());
 
@@ -118,6 +118,16 @@ public Document toDocument(AggregationOperationContext context) {
 		return new Document(getOperator(), graphLookup);
 	}
 
+	String getCollectionName(AggregationOperationContext context) {
+		return from instanceof Class<?> type ? context.getCollection(type) : from.toString();
+	}
+
+	String getForeignFieldName(AggregationOperationContext context) {
+
+		return from instanceof Class<?> type ? context.getMappedFieldName(type, connectFrom.getTarget())
+				: connectFrom.getTarget();
+	}
+
 	@Override
 	public String getOperator() {
 		return "$graphLookup";
@@ -128,7 +138,7 @@ public ExposedFields getFields() {
 
 		List<ExposedField> fields = new ArrayList<>(2);
 		fields.add(new ExposedField(as, true));
-		if(depthField != null) {
+		if (depthField != null) {
 			fields.add(new ExposedField(depthField, true));
 		}
 		return ExposedFields.from(fields.toArray(new ExposedField[0]));
@@ -146,6 +156,17 @@ public interface FromBuilder {
 		 * @return never {@literal null}.
 		 */
 		StartWithBuilder from(String collectionName);
+
+		/**
+		 * Use the given type to determine name of the foreign collection and map
+		 * {@link ConnectFromBuilder#connectFrom(String)} against it to consider eventually present
+		 * {@link org.springframework.data.mongodb.core.mapping.Field} annotations.
+		 *
+		 * @param type must not be {@literal null}.
+		 * @return never {@literal null}.
+		 * @since 4.2
+		 */
+		StartWithBuilder from(Class<?> type);
 	}
 
 	/**
@@ -218,7 +239,7 @@ public interface ConnectToBuilder {
 	static final class GraphLookupOperationFromBuilder
 			implements FromBuilder, StartWithBuilder, ConnectFromBuilder, ConnectToBuilder {
 
-		private @Nullable String from;
+		private @Nullable Object from;
 		private @Nullable List<? extends Object> startWith;
 		private @Nullable String connectFrom;
 
@@ -231,6 +252,14 @@ public StartWithBuilder from(String collectionName) {
 			return this;
 		}
 
+		@Override
+		public StartWithBuilder from(Class<?> type) {
+
+			Assert.notNull(type, "Type must not be null");
+			this.from = type;
+			return this;
+		}
+
 		@Override
 		public ConnectFromBuilder startWith(String... fieldReferences) {
 
@@ -321,15 +350,15 @@ public GraphLookupOperationBuilder connectTo(String fieldName) {
 	 */
 	public static final class GraphLookupOperationBuilder {
 
-		private final String from;
+		private final Object from;
 		private final List<Object> startWith;
 		private final Field connectFrom;
 		private final Field connectTo;
 		private @Nullable Long maxDepth;
 		private @Nullable Field depthField;
 		private @Nullable CriteriaDefinition restrictSearchWithMatch;
 
-		protected GraphLookupOperationBuilder(String from, List<? extends Object> startWith, String connectFrom,
+		protected GraphLookupOperationBuilder(Object from, List<? extends Object> startWith, String connectFrom,
 				String connectTo) {
 
 			this.from = from;

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/LookupOperation.java

@@ -39,7 +39,7 @@
  */
 public class LookupOperation implements FieldsExposingAggregationOperation, InheritsFieldsAggregationOperation {
 
-	private final String from;
+	private Object from;
 
 	@Nullable //
 	private final Field localField;
@@ -97,6 +97,22 @@ public LookupOperation(String from, @Nullable Let let, AggregationPipeline pipel
 	 */
 	public LookupOperation(String from, @Nullable Field localField, @Nullable Field foreignField, @Nullable Let let,
 			@Nullable AggregationPipeline pipeline, Field as) {
+		this((Object) from, localField, foreignField, let, pipeline, as);
+	}
+
+	/**
+	 * Creates a new {@link LookupOperation} for the given combination of {@link Field}s and {@link AggregationPipeline
+	 * pipeline}.
+	 *
+	 * @param from must not be {@literal null}. Can be eiter the target collection name or a {@link Class}.
+	 * @param localField can be {@literal null} if {@literal pipeline} is present.
+	 * @param foreignField can be {@literal null} if {@literal pipeline} is present.
+	 * @param let can be {@literal null} if {@literal localField} and {@literal foreignField} are present.
+	 * @param as must not be {@literal null}.
+	 * @since 4.2
+	 */
+	private LookupOperation(Object from, @Nullable Field localField, @Nullable Field foreignField, @Nullable Let let,
+			@Nullable AggregationPipeline pipeline, Field as) {
 
 		Assert.notNull(from, "From must not be null");
 		if (pipeline == null) {
@@ -125,12 +141,14 @@ public Document toDocument(AggregationOperationContext context) {
 
 		Document lookupObject = new Document();
 
-		lookupObject.append("from", from);
+		lookupObject.append("from", getCollectionName(context));
+
 		if (localField != null) {
 			lookupObject.append("localField", localField.getTarget());
 		}
+
 		if (foreignField != null) {
-			lookupObject.append("foreignField", foreignField.getTarget());
+			lookupObject.append("foreignField", getForeignFieldName(context));
 		}
 		if (let != null) {
 			lookupObject.append("let", let.toDocument(context).get("$let", Document.class).get("vars"));
@@ -144,6 +162,16 @@ public Document toDocument(AggregationOperationContext context) {
 		return new Document(getOperator(), lookupObject);
 	}
 
+	String getCollectionName(AggregationOperationContext context) {
+		return from instanceof Class<?> type ? context.getCollection(type) : from.toString();
+	}
+
+	String getForeignFieldName(AggregationOperationContext context) {
+
+		return from instanceof Class<?> type ? context.getMappedFieldName(type, foreignField.getTarget())
+				: foreignField.getTarget();
+	}
+
 	@Override
 	public String getOperator() {
 		return "$lookup";
@@ -158,16 +186,28 @@ public static FromBuilder newLookup() {
 		return new LookupOperationBuilder();
 	}
 
-	public static interface FromBuilder {
+	public interface FromBuilder {
 
 		/**
 		 * @param name the collection in the same database to perform the join with, must not be {@literal null} or empty.
 		 * @return never {@literal null}.
 		 */
 		LocalFieldBuilder from(String name);
+
+		/**
+		 * Use the given type to determine name of the foreign collection and map
+		 * {@link ForeignFieldBuilder#foreignField(String)} against it to consider eventually present
+		 * {@link org.springframework.data.mongodb.core.mapping.Field} annotations.
+		 *
+		 * @param type the type of the target collection in the same database to perform the join with, must not be
+		 *          {@literal null}.
+		 * @return never {@literal null}.
+		 * @since 4.2
+		 */
+		LocalFieldBuilder from(Class<?> type);
 	}
 
-	public static interface LocalFieldBuilder extends PipelineBuilder {
+	public interface LocalFieldBuilder extends PipelineBuilder {
 
 		/**
 		 * @param name the field from the documents input to the {@code $lookup} stage, must not be {@literal null} or
@@ -177,7 +217,7 @@ public static interface LocalFieldBuilder extends PipelineBuilder {
 		ForeignFieldBuilder localField(String name);
 	}
 
-	public static interface ForeignFieldBuilder {
+	public interface ForeignFieldBuilder {
 
 		/**
 		 * @param name the field from the documents in the {@code from} collection, must not be {@literal null} or empty.
@@ -246,7 +286,7 @@ default AsBuilder pipeline(AggregationOperation... stages) {
 		LookupOperation as(String name);
 	}
 
-	public static interface AsBuilder extends PipelineBuilder {
+	public interface AsBuilder extends PipelineBuilder {
 
 		/**
 		 * @param name the name of the new array field to add to the input documents, must not be {@literal null} or empty.
@@ -264,7 +304,7 @@ public static interface AsBuilder extends PipelineBuilder {
 	public static final class LookupOperationBuilder
 			implements FromBuilder, LocalFieldBuilder, ForeignFieldBuilder, AsBuilder {
 
-		private @Nullable String from;
+		private @Nullable Object from;
 		private @Nullable Field localField;
 		private @Nullable Field foreignField;
 		private @Nullable ExposedField as;
@@ -288,6 +328,14 @@ public LocalFieldBuilder from(String name) {
 			return this;
 		}
 
+		@Override
+		public LocalFieldBuilder from(Class<?> type) {
+
+			Assert.notNull(type, "'From' must not be null");
+			from = type;
+			return this;
+		}
+
 		@Override
 		public AsBuilder foreignField(String name) {
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/PrefixingDelegatingAggregationOperationContext.java

@@ -30,9 +30,8 @@
 
 /**
  * {@link AggregationOperationContext} implementation prefixing non-command keys on root level with the given prefix.
- * Useful when mapping fields to domain specific types while having to prefix keys for query purpose.
- * <br />
- * Fields to be excluded from prefixing my be added to a {@literal denylist}.
+ * Useful when mapping fields to domain specific types while having to prefix keys for query purpose. <br />
+ * Fields to be excluded from prefixing can be added to a {@literal denylist}.
  *
  * @author Christoph Strobl
  * @author Mark Paluch

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/TypeBasedAggregationOperationContext.java

@@ -92,6 +92,20 @@ public FieldReference getReference(String name) {
 		return getReferenceFor(field(name));
 	}
 
+	@Override
+	public String getCollection(Class<?> type) {
+
+		MongoPersistentEntity<?> persistentEntity = mappingContext.getPersistentEntity(type);
+		return persistentEntity != null ? persistentEntity.getCollection() : AggregationOperationContext.super.getCollection(type);
+	}
+
+	@Override
+	public String getMappedFieldName(Class<?> type, String field) {
+
+		PersistentPropertyPath<MongoPersistentProperty> persistentPropertyPath = mappingContext.getPersistentPropertyPath(field, type);
+		return persistentPropertyPath.getLeafProperty().getFieldName();
+	}
+
 	@Override
 	public Fields getFields(Class<?> type) {