Add support for $bottom aggregation operator.

Closes #4139
Original pull request: #4182.

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

@@ -22,8 +22,12 @@ import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import org.bson.Document;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Order;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
@@ -68,8 +72,24 @@ abstract class AbstractAggregationExpression implements AggregationExpression {
 			return ((AggregationExpression) value).toDocument(context);
 		}
 
-		if (value instanceof Field) {
-			return context.getReference((Field) value).toString();
+		if (value instanceof Field field) {
+			return context.getReference(field).toString();
+		}
+
+		if(value instanceof Fields fields) {
+			return fields.asList().stream().map(it -> unpack(it, context)).collect(Collectors.toList());
+		}
+
+		if(value instanceof Sort sort) {
+
+			Document sortDoc = new Document();
+			for (Order order : sort) {
+
+				// Check reference
+				FieldReference reference = context.getReference(order.getProperty());
+				sortDoc.put(reference.getRaw(), order.isAscending() ? 1 : -1);
+			}
+			return sortDoc;
 		}
 
 		if (value instanceof List) {

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

@@ -387,6 +387,17 @@ public class GroupOperation implements FieldsExposingAggregationOperation {
 		return new GroupOperationBuilder(this, new Operation(accumulator));
 	}
 
+	/**
+	 * Adds a computed field to the {@link GroupOperation}.
+	 *
+	 * @param expression must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @since 4.0
+	 */
+	public GroupOperation and(String fieldName, AggregationExpression expression) {
+		return new GroupOperationBuilder(this, new Operation(expression)).as(fieldName);
+	}
+
 	private GroupOperationBuilder newBuilder(Keyword keyword, @Nullable String reference, @Nullable Object value) {
 		return new GroupOperationBuilder(this, new Operation(keyword, null, reference, value));
 	}

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import org.springframework.data.domain.Sort;
+
+/**
+ * Gateway to {@literal selection operators} such as {@literal $bottom}.
+ *
+ * @author Christoph Strobl
+ * @since 4.0
+ */
+public class SelectionOperators {
+
+	/**
+	 * {@link AbstractAggregationExpression} to return the bottom element according to the specified {@link #sortBy(Sort)
+	 * order}.
+	 */
+	public static class Bottom extends AbstractAggregationExpression {
+
+		private Bottom(Object value) {
+			super(value);
+		}
+
+		@Override
+		protected String getMongoMethod() {
+			return "$bottom";
+		}
+
+		/**
+		 * @return new instance of {@link Bottom}.
+		 */
+		public static Bottom bottom() {
+			return new Bottom(Collections.emptyMap());
+		}
+
+		/**
+		 * Define result ordering.
+		 *
+		 * @param sort must not be {@literal null}.
+		 * @return new instance of {@link Bottom}.
+		 */
+		public Bottom sortBy(Sort sort) {
+			return new Bottom(append("sortBy", sort));
+		}
+
+		/**
+		 * Define result ordering.
+		 *
+		 * @param out must not be {@literal null}.
+		 * @return new instance of {@link Bottom}.
+		 */
+		public Bottom output(Fields out) {
+			return new Bottom(append("output", out));
+		}
+
+		/**
+		 * Define fields included in the output for each element.
+		 *
+		 * @param fieldNames must not be {@literal null}.
+		 * @return new instance of {@link Bottom}.
+		 * @see #output(Fields)
+		 */
+		public Bottom output(String... fieldNames) {
+			return output(Fields.fields(fieldNames));
+		}
+
+		/**
+		 * Define expressions building the value included in the output for each element.
+		 *
+		 * @param out must not be {@literal null}.
+		 * @return new instance of {@link Bottom}.
+		 * @see #output(Fields)
+		 */
+		public Bottom output(AggregationExpression... out) {
+			return new Bottom(append("output", Arrays.asList(out)));
+		}
+	}
+}

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/spel/MethodReferenceNode.java

@@ -226,6 +226,10 @@ public class MethodReferenceNode extends ExpressionNode {
 		map.put("toString", singleArgRef().forOperator("$toString"));
 		map.put("degreesToRadians", singleArgRef().forOperator("$degreesToRadians"));
 
+		// SELECT OPERATORS
+		map.put("bottom", mapArgRef().forOperator("$bottom") //
+				.mappingParametersTo("output", "sortBy"));
+
 		FUNCTIONS = Collections.unmodifiableMap(map);
 	}
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/GroupOperationUnitTests.java

@@ -22,7 +22,10 @@ import java.util.Arrays;
 
 import org.bson.Document;
 import org.junit.jupiter.api.Test;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mongodb.core.DocumentTestUtils;
+import org.springframework.data.mongodb.core.aggregation.SelectionOperators.Bottom;
 import org.springframework.data.mongodb.core.query.Criteria;
 
 /**
@@ -252,6 +255,17 @@ class GroupOperationUnitTests {
 		assertThat(accumulatedValue).containsKey("$accumulator");
 	}
 
+	@Test // GH-4139
+	void groupOperationAllowsToAddFieldsComputedViaExpression() {
+
+		GroupOperation groupOperation = Aggregation.group("id").and("playerId",
+				Bottom.bottom().output("playerId", "score").sortBy(Sort.by(Direction.DESC, "score")));
+		Document groupClause = extractDocumentFromGroupOperation(groupOperation);
+
+		assertThat(groupClause).containsEntry("playerId",
+				Document.parse("{ $bottom : { output: [ \"$playerId\", \"$score\" ], sortBy: { \"score\": -1 }}}"));
+	}
+
 	private Document extractDocumentFromGroupOperation(GroupOperation groupOperation) {
 		Document document = groupOperation.toDocument(Aggregation.DEFAULT_CONTEXT);
 		Document groupClause = DocumentTestUtils.getAsDocument(document, "$group");

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/SelectionOperatorUnitTests.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import static org.assertj.core.api.Assertions.*;
+
+import org.bson.Document;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.domain.Sort;
+import org.springframework.data.domain.Sort.Direction;
+import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
+import org.springframework.data.mongodb.core.convert.NoOpDbRefResolver;
+import org.springframework.data.mongodb.core.convert.QueryMapper;
+import org.springframework.data.mongodb.core.mapping.Field;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+
+/**
+ * @author Christoph Strobl
+ */
+class SelectionOperatorUnitTests {
+
+	@Test // GH-4139
+	void bottomRenderedCorrectly() {
+
+		Document document = SelectionOperators.Bottom.bottom().output(Fields.fields("playerId", "score"))
+				.sortBy(Sort.by(Direction.DESC, "score")).toDocument(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(document).isEqualTo(Document.parse("""
+				{
+				   $bottom:
+				   {
+				      output: [ "$playerId", "$score" ],
+				      sortBy: { "score": -1 }
+				   }
+				}
+				"""));
+	}
+
+	@Test // GH-4139
+	void bottomMapsFieldNamesCorrectly() {
+
+		MongoMappingContext mappingContext = new MongoMappingContext();
+		RelaxedTypeBasedAggregationOperationContext aggregationContext = new RelaxedTypeBasedAggregationOperationContext(
+				Player.class, mappingContext,
+				new QueryMapper(new MappingMongoConverter(NoOpDbRefResolver.INSTANCE, mappingContext)));
+
+		Document document = SelectionOperators.Bottom.bottom().output(Fields.fields("playerId", "score"))
+				.sortBy(Sort.by(Direction.DESC, "score")).toDocument(aggregationContext);
+
+		assertThat(document).isEqualTo(Document.parse("""
+				{
+				   $bottom:
+				   {
+				      output: [ "$player_id", "$s_cor_e" ],
+				      sortBy: { "s_cor_e": -1 }
+				   }
+				}
+				"""));
+	}
+
+	static class Player {
+
+		@Field("player_id") String playerId;
+
+		@Field("s_cor_e") Integer score;
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/SpelExpressionTransformerUnitTests.java

@@ -1174,6 +1174,11 @@ public class SpelExpressionTransformerUnitTests {
 		assertThat(transform("rand()")).isEqualTo("{ $rand : {} }");
 	}
 
+	@Test // GH-4139
+	void shouldRenderBottom() {
+		assertThat(transform("bottom(new String[]{\"$playerId\", \"$score\" }, { \"score\" : -1 })")).isEqualTo("{ $bottom : { output: [ \"$playerId\", \"$score\" ], sortBy: { \"score\": -1 }}}");
+	}
+
 	private Document transform(String expression, Object... params) {
 		return (Document) transformer.transform(expression, Aggregation.DEFAULT_CONTEXT, params);
 	}

src/main/asciidoc/reference/aggregation-framework.adoc

@@ -120,6 +120,9 @@ At the time of this writing, we provide support for the following Aggregation Op
 | Script Aggregation Operators
 | `function`, `accumulator`
 
+| Selection Aggregation Operators
+| `bottom`
+
 |===
 
 +++*+++ The operation is mapped or added by Spring Data MongoDB.