Polishing.

Reuse Let from VariableOperators.
Limit API exposure and favor builders.
Update nullability constraints and assertions.
Update integration tests.
Add unit tests.

Original Pull Request: #4272

Author: christophstrobl

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

@@ -28,6 +28,7 @@
 import org.springframework.data.mongodb.core.aggregation.CountOperation.CountOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.FacetOperation.FacetOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.StartWithBuilder;
+import org.springframework.data.mongodb.core.aggregation.LookupOperation.LookupOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.MergeOperation.MergeOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.ReplaceRootOperation.ReplaceRootDocumentOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.ReplaceRootOperation.ReplaceRootOperationBuilder;
@@ -665,16 +666,21 @@ public static LookupOperation lookup(Field from, Field localField, Field foreign
 		return new LookupOperation(from, localField, foreignField, as);
 	}
 
-	public static LookupOperation lookup(String from, String localField, String foreignField, String as, List<AggregationOperation> aggregationOperations) {
-		return lookup(field(from), field(localField), field(foreignField), field(as), null, new AggregationPipeline(aggregationOperations));
-	}
-
-	public static LookupOperation lookup(String from, String localField, String foreignField, String as, List<LookupOperation.Let.ExpressionVariable> letExpressionVars, List<AggregationOperation> aggregationOperations) {
-		return lookup(field(from), field(localField), field(foreignField), field(as), new LookupOperation.Let(letExpressionVars), new AggregationPipeline(aggregationOperations));
-	}
-
-	public static LookupOperation lookup(Field from, Field localField, Field foreignField, Field as, LookupOperation.Let let, AggregationPipeline pipeline) {
-		return new LookupOperation(from, localField, foreignField, as, let, pipeline);
+	/**
+	 * Entrypoint for creating {@link LookupOperation $lookup} using a fluent builder API.
+	 * <pre class="code">
+	 * Aggregation.lookup().from("restaurants")
+	 * 	.localField("restaurant_name")
+	 * 	.foreignField("name")
+	 * 	.let(newVariable("orders_drink").forField("drink"))
+	 * 	.pipeline(match(ctx -> new Document("$expr", new Document("$in", List.of("$$orders_drink", "$beverages")))))
+	 * 	.as("matches")
+	 * </pre>
+	 * @return new instance of {@link LookupOperationBuilder}.
+	 * @since 4.1
+	 */
+	public static LookupOperationBuilder lookup() {
+		return new LookupOperationBuilder();
 	}
 
 	/**

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

@@ -15,17 +15,19 @@
  */
 package org.springframework.data.mongodb.core.aggregation;
 
-import java.util.List;
+import java.util.function.Supplier;
 
 import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.VariableOperators.Let;
+import org.springframework.data.mongodb.core.aggregation.VariableOperators.Let.ExpressionVariable;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
- * Encapsulates the aggregation framework {@code $lookup}-operation. We recommend to use the static factory method
- * {@link Aggregation#lookup(String, String, String, String)} instead of creating instances of this class directly.
+ * Encapsulates the aggregation framework {@code $lookup}-operation. We recommend to use the builder provided via
+ * {@link #newLookup()} instead of creating instances of this class directly.
  *
  * @author Alessio Fachechi
  * @author Christoph Strobl
@@ -37,16 +39,22 @@
  */
 public class LookupOperation implements FieldsExposingAggregationOperation, InheritsFieldsAggregationOperation {
 
-	private final Field from;
+	private final String from;
+
+	@Nullable //
 	private final Field localField;
+
+	@Nullable //
 	private final Field foreignField;
-	private final ExposedField as;
 
-	@Nullable
+	@Nullable //
 	private final Let let;
-	@Nullable
+
+	@Nullable //
 	private final AggregationPipeline pipeline;
 
+	private final ExposedField as;
+
 	/**
 	 * Creates a new {@link LookupOperation} for the given {@link Field}s.
 	 *
@@ -56,13 +64,47 @@ public class LookupOperation implements FieldsExposingAggregationOperation, Inhe
 	 * @param as must not be {@literal null}.
 	 */
 	public LookupOperation(Field from, Field localField, Field foreignField, Field as) {
-		this(from, localField, foreignField, as, null, null);
+		this(((Supplier<String>) () -> {
+
+			Assert.notNull(from, "From must not be null");
+			return from.getTarget();
+		}).get(), localField, foreignField, null, null, as);
 	}
 
-	public LookupOperation(Field from, Field localField, Field foreignField, Field as, @Nullable Let let, @Nullable AggregationPipeline pipeline) {
+	/**
+	 * Creates a new {@link LookupOperation} for the given combination of {@link Field}s and {@link AggregationPipeline
+	 * pipeline}.
+	 *
+	 * @param from must not be {@literal null}.
+	 * @param let must not be {@literal null}.
+	 * @param as must not be {@literal null}.
+	 * @since 4.1
+	 */
+	public LookupOperation(String from, @Nullable Let let, AggregationPipeline pipeline, Field as) {
+		this(from, null, null, 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}.
+	 * @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.1
+	 */
+	public LookupOperation(String from, @Nullable Field localField, @Nullable Field foreignField, @Nullable Let let,
+			@Nullable AggregationPipeline pipeline, Field as) {
+
 		Assert.notNull(from, "From must not be null");
-		Assert.notNull(localField, "LocalField must not be null");
-		Assert.notNull(foreignField, "ForeignField must not be null");
+		if (pipeline == null) {
+			Assert.notNull(localField, "LocalField must not be null");
+			Assert.notNull(foreignField, "ForeignField must not be null");
+		} else if (localField == null && foreignField == null) {
+			Assert.notNull(pipeline, "Pipeline must not be null");
+		}
 		Assert.notNull(as, "As must not be null");
 
 		this.from = from;
@@ -83,19 +125,22 @@ public Document toDocument(AggregationOperationContext context) {
 
 		Document lookupObject = new Document();
 
-		lookupObject.append("from", from.getTarget());
-		lookupObject.append("localField", localField.getTarget());
-		lookupObject.append("foreignField", foreignField.getTarget());
-		lookupObject.append("as", as.getTarget());
-
+		lookupObject.append("from", from);
+		if (localField != null) {
+			lookupObject.append("localField", localField.getTarget());
+		}
+		if (foreignField != null) {
+			lookupObject.append("foreignField", foreignField.getTarget());
+		}
 		if (let != null) {
-			lookupObject.append("let", let.toDocument(context));
+			lookupObject.append("let", let.toDocument(context).get("$let", Document.class).get("vars"));
 		}
-
 		if (pipeline != null) {
 			lookupObject.append("pipeline", pipeline.toDocuments(context));
 		}
 
+		lookupObject.append("as", as.getTarget());
+
 		return new Document(getOperator(), lookupObject);
 	}
 
@@ -122,7 +167,7 @@ public static interface FromBuilder {
 		LocalFieldBuilder from(String name);
 	}
 
-	public static interface LocalFieldBuilder {
+	public static interface LocalFieldBuilder extends PipelineBuilder {
 
 		/**
 		 * @param name the field from the documents input to the {@code $lookup} stage, must not be {@literal null} or
@@ -141,7 +186,67 @@ public static interface ForeignFieldBuilder {
 		AsBuilder foreignField(String name);
 	}
 
-	public static interface AsBuilder {
+	/**
+	 * @since 4.1
+	 * @author Christoph Strobl
+	 */
+	public interface LetBuilder {
+
+		/**
+		 * Specifies {@link Let#getVariableNames() variables) that can be used in the
+		 * {@link PipelineBuilder#pipeline(AggregationOperation...) pipeline stages}.
+		 *
+		 * @param let must not be {@literal null}.
+		 * @return never {@literal null}.
+		 * @see PipelineBuilder
+		 */
+		PipelineBuilder let(Let let);
+
+		/**
+		 * Specifies {@link Let#getVariableNames() variables) that can be used in the
+		 * {@link PipelineBuilder#pipeline(AggregationOperation...) pipeline stages}.
+		 *
+		 * @param variables must not be {@literal null}.
+		 * @return never {@literal null}.
+		 * @see PipelineBuilder
+		 */
+		default PipelineBuilder let(ExpressionVariable... variables) {
+			return let(Let.just(variables));
+		}
+	}
+
+	/**
+	 * @since 4.1
+	 * @author Christoph Strobl
+	 */
+	public interface PipelineBuilder extends LetBuilder {
+
+		/**
+		 * Specifies the {@link AggregationPipeline pipeline} that determines the resulting documents.
+		 *
+		 * @param pipeline must not be {@literal null}.
+		 * @return never {@literal null}.
+		 */
+		AsBuilder pipeline(AggregationPipeline pipeline);
+
+		/**
+		 * Specifies the {@link AggregationPipeline#getOperations() stages} that determine the resulting documents.
+		 *
+		 * @param stages must not be {@literal null} can be empty.
+		 * @return never {@literal null}.
+		 */
+		default AsBuilder pipeline(AggregationOperation... stages) {
+			return pipeline(AggregationPipeline.of(stages));
+		}
+
+		/**
+		 * @param name the name of the new array field to add to the input documents, must not be {@literal null} or empty.
+		 * @return new instance of {@link LookupOperation}.
+		 */
+		LookupOperation as(String name);
+	}
+
+	public static 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.
@@ -159,10 +264,12 @@ public static interface AsBuilder {
 	public static final class LookupOperationBuilder
 			implements FromBuilder, LocalFieldBuilder, ForeignFieldBuilder, AsBuilder {
 
-		private @Nullable Field from;
+		private @Nullable String from;
 		private @Nullable Field localField;
 		private @Nullable Field foreignField;
 		private @Nullable ExposedField as;
+		private @Nullable Let let;
+		private @Nullable AggregationPipeline pipeline;
 
 		/**
 		 * Creates new builder for {@link LookupOperation}.
@@ -177,18 +284,10 @@ public static FromBuilder newBuilder() {
 		public LocalFieldBuilder from(String name) {
 
 			Assert.hasText(name, "'From' must not be null or empty");
-			from = Fields.field(name);
+			from = name;
 			return this;
 		}
 
-		@Override
-		public LookupOperation as(String name) {
-
-			Assert.hasText(name, "'As' must not be null or empty");
-			as = new ExposedField(Fields.field(name), true);
-			return new LookupOperation(from, localField, foreignField, as);
-		}
-
 		@Override
 		public AsBuilder foreignField(String name) {
 
@@ -204,50 +303,29 @@ public ForeignFieldBuilder localField(String name) {
 			localField = Fields.field(name);
 			return this;
 		}
-	}
-
-	public static class Let implements AggregationExpression{
-
-		private final List<ExpressionVariable> vars;
-
-		public Let(List<ExpressionVariable> vars) {
-			Assert.notEmpty(vars, "'let' must not be null or empty");
-			this.vars = vars;
-		}
 
 		@Override
-		public Document toDocument(AggregationOperationContext context) {
-			return toLet();
-		}
-
-		private Document toLet() {
-			Document mappedVars = new Document();
+		public PipelineBuilder let(Let let) {
 
-			for (ExpressionVariable var : this.vars) {
-				mappedVars.putAll(getMappedVariable(var));
-			}
-
-			return mappedVars;
+			Assert.notNull(let, "Let must not be null");
+			this.let = let;
+			return this;
 		}
 
-		private Document getMappedVariable(ExpressionVariable var) {
-			return new Document(var.variableName, prefixDollarSign(var.expression));
-		}
+		@Override
+		public AsBuilder pipeline(AggregationPipeline pipeline) {
 
-		private String prefixDollarSign(String expression) {
-			return "$" + expression;
+			Assert.notNull(pipeline, "Pipeline must not be null");
+			this.pipeline = pipeline;
+			return this;
 		}
 
-		public static class ExpressionVariable {
-
-			private final String variableName;
-
-			private final String expression;
+		@Override
+		public LookupOperation as(String name) {
 
-			public ExpressionVariable(String variableName, String expression) {
-				this.variableName = variableName;
-				this.expression = expression;
-			}
+			Assert.hasText(name, "'As' must not be null or empty");
+			as = new ExposedField(Fields.field(name), true);
+			return new LookupOperation(from, localField, foreignField, let, pipeline, as);
 		}
 	}
 }