Add support for $bottomN aggregation operator.

Closes #4139
Original pull request: #4182.

Author: christophstrobl

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

@@ -38,9 +38,14 @@ private Bottom(Object value) {
 			super(value);
 		}
 
+		/**
+		 * In case a limit value ({@literal n}) is present {@literal $bottomN} is used instead of {@literal $bottom}.
+		 *
+		 * @return
+		 */
 		@Override
 		protected String getMongoMethod() {
-			return "$bottom";
+			return get("n") == null ? "$bottom" : "$bottomN";
 		}
 
 		/**
@@ -50,6 +55,31 @@ public static Bottom bottom() {
 			return new Bottom(Collections.emptyMap());
 		}
 
+		/**
+		 * Limits the number of returned elements to the given value.
+		 *
+		 * @param numberOfResults
+		 * @return new instance of {@link Bottom}.
+		 */
+		public Bottom limit(int numberOfResults) {
+			return limit((Object) numberOfResults);
+		}
+
+		/**
+		 * Limits the number of returned elements to the value defined by the given {@link AggregationExpression
+		 * expression}.
+		 *
+		 * @param expression must not be {@literal null}.
+		 * @return new instance of {@link Bottom}.
+		 */
+		public Bottom limit(AggregationExpression expression) {
+			return limit((Object) expression);
+		}
+
+		private Bottom limit(Object value) {
+			return new Bottom(append("n", value));
+		}
+
 		/**
 		 * Define result ordering.
 		 *

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

@@ -212,6 +212,10 @@ public class MethodReferenceNode extends ExpressionNode {
 		// OBJECT OPERATORS
 		map.put("objectToArray", singleArgRef().forOperator("$objectToArray"));
 		map.put("mergeObjects", arrayArgRef().forOperator("$mergeObjects"));
+		map.put("bottom", mapArgRef().forOperator("$bottom") //
+				.mappingParametersTo("output", "sortBy"));
+		map.put("bottomN", mapArgRef().forOperator("$bottomN") //
+				.mappingParametersTo("n", "output", "sortBy"));
 
 		// CONVERT OPERATORS
 		map.put("convert", mapArgRef().forOperator("$convert") //
@@ -226,9 +230,6 @@ 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);
 	}

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

@@ -71,6 +71,24 @@ void bottomMapsFieldNamesCorrectly() {
 				"""));
 	}
 
+	@Test // GH-4139
+	void bottomNRenderedCorrectly() {
+
+		Document document = SelectionOperators.Bottom.bottom().output(Fields.fields("playerId", "score"))
+				.sortBy(Sort.by(Direction.DESC, "score")).limit(3).toDocument(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(document).isEqualTo(Document.parse("""
+				{
+				   $bottomN:
+				   {
+				      n : 3,
+				      output: [ "$playerId", "$score" ],
+				      sortBy: { "score": -1 }
+				   }
+				}
+				"""));
+	}
+
 	static class Player {
 
 		@Field("player_id") String playerId;

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

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

Diff for: src/main/asciidoc/reference/aggregation-framework.adoc

@@ -82,7 +82,7 @@ At the time of this writing, we provide support for the following Aggregation Op
 | `setEquals`, `setIntersection`, `setUnion`, `setDifference`, `setIsSubset`, `anyElementTrue`, `allElementsTrue`
 
 | Group/Accumulator Aggregation Operators
-| `addToSet`, `covariancePop`, `covarianceSamp`, `expMovingAvg`, `first`, `last`, `max`, `min`, `avg`, `push`, `sum`, `count` (+++*+++), `stdDevPop`, `stdDevSamp`
+| `addToSet`, `bottom`, `bottomN`, `covariancePop`, `covarianceSamp`, `expMovingAvg`, `first`, `last`, `max`, `min`, `avg`, `push`, `sum`, `count` (+++*+++), `stdDevPop`, `stdDevSamp`
 
 | Arithmetic Aggregation Operators
 | `abs`, `acos`, `acosh`, `add` (+++*+++ via `plus`), `asin`, `asin`, `atan`, `atan2`, `atanh`, `ceil`, `cos`, `cosh`, `derivative`, `divide`, `exp`, `floor`, `integral`, `ln`, `log`, `log10`, `mod`, `multiply`, `pow`, `round`, `sqrt`, `subtract` (+++*+++ via `minus`), `sin`, `sinh`, `tan`, `tanh`, `trunc`
@@ -120,9 +120,6 @@ 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.